``` ├── .github/ ├── ISSUE_TEMPLATE/ ├── bug_report.md ├── feature_request.md ├── PULL_REQUEST_TEMPLATE.md ├── static/ ├── img/ ├── eino/ ├── chain.png ├── chain_append_branch.png ├── chain_append_parallel.png ├── eino_concept.jpeg ├── eino_framework.jpeg ├── graph.gif ├── graph.png ├── graph_add_branch.png ├── graph_add_edge.png ├── graph_parallel.png ├── lark_group_en.png ├── lark_group_zh.png ├── react.png ├── simple_chain.png ├── stream.png ├── tool_call_graph.png ├── workflows/ ├── pr-check.yml ├── tests.yml ├── .gitignore ├── .golangci.yaml ├── .licenserc.yaml ├── .testcoverage.yml ├── CODE_OF_CONDUCT.md ├── CONTRIBUTING.md ├── LICENSE-APACHE ├── README.md ├── README.zh_CN.md ├── _typos.toml ├── callbacks/ ├── aspect_inject.go ├── aspect_inject_test.go ├── doc.go ├── handler_builder.go ├── interface.go ├── interface_test.go ├── components/ ├── document/ ├── callback_extra_loader.go ├── callback_extra_transformer.go ├── doc.go ├── interface.go ├── option.go ├── option_test.go ├── parser/ ├── ext_parser.go ├── interface.go ├── option.go ├── option_test.go ├── parser_test.go ├── testdata/ ├── test.md ├── text_parser.go ├── embedding/ ├── callback_extra.go ├── callback_extra_test.go ├── doc.go ├── interface.go ├── option.go ├── option_test.go ├── indexer/ ├── callback_extra.go ├── callback_extra_test.go ├── doc.go ├── interface.go ├── option.go ├── option_test.go ├── model/ ├── callback_extra.go ├── callback_extra_test.go ├── doc.go ├── interface.go ├── option.go ├── option_test.go ├── prompt/ ├── callback_extra.go ├── callback_extra_test.go ├── chat_template.go ├── chat_template_test.go ├── doc.go ├── interface.go ├── option.go ├── option_test.go ├── retriever/ ├── callback_extra.go ├── callback_extra_test.go ├── doc.go ├── interface.go ├── option.go ├── option_test.go ├── tool/ ├── callback_extra.go ├── callback_extra_test.go ├── doc.go ├── interface.go ├── option.go ├── option_test.go ├── utils/ ├── create_options.go ├── doc.go ├── error_handler.go ├── error_handler_test.go ├── invokable_func.go ├── invokable_func_test.go ├── streamable_func.go ├── streamable_func_test.go ├── types.go ├── compose/ ├── branch.go ├── branch_test.go ├── chain.go ├── chain_branch.go ``` ## /.github/ISSUE_TEMPLATE/bug_report.md --- name: Bug report about: Create a report to help us improve title: '' labels: '' assignees: '' --- **Describe the bug** A clear and concise description of what the bug is. **To Reproduce** Steps to reproduce the behavior: 1. Go to '...' 2. Click on '....' 3. Scroll down to '....' 4. See error **Expected behavior** A clear and concise description of what you expected to happen. **Screenshots** If applicable, add screenshots to help explain your problem. **Version:** Please provide the version of {project_name} you are using. **Environment:** The output of `go env`. **Additional context** Add any other context about the problem here. ## /.github/ISSUE_TEMPLATE/feature_request.md --- name: Feature request about: Suggest an idea for this project title: '' labels: '' assignees: '' --- **Is your feature request related to a problem? Please describe.** A clear and concise description of what the problem is. Ex. I'm always frustrated when [...] **Describe the solution you'd like** A clear and concise description of what you want to happen. **Describe alternatives you've considered** A clear and concise description of any alternative solutions or features you've considered. **Additional context** Add any other context or screenshots about the feature request here. ## /.github/PULL_REQUEST_TEMPLATE.md #### What type of PR is this? #### Check the PR title. - [ ] This PR title match the format: \(optional scope): \ - [ ] The description of this PR title is user-oriented and clear enough for others to understand. - [ ] Attach the PR updating the user documentation if the current PR requires user awareness at the usage level. [User docs repo](https://github.com/cloudwego/cloudwego.github.io) #### (Optional) Translate the PR title into Chinese. #### (Optional) More detailed description for this PR(en: English/zh: Chinese). en: zh(optional): #### (Optional) Which issue(s) this PR fixes: #### (optional) The PR that updates user documentation: ## /.github/static/img/eino/chain.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/chain.png ## /.github/static/img/eino/chain_append_branch.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/chain_append_branch.png ## /.github/static/img/eino/chain_append_parallel.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/chain_append_parallel.png ## /.github/static/img/eino/eino_concept.jpeg Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/eino_concept.jpeg ## /.github/static/img/eino/eino_framework.jpeg Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/eino_framework.jpeg ## /.github/static/img/eino/graph.gif Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/graph.gif ## /.github/static/img/eino/graph.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/graph.png ## /.github/static/img/eino/graph_add_branch.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/graph_add_branch.png ## /.github/static/img/eino/graph_add_edge.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/graph_add_edge.png ## /.github/static/img/eino/graph_parallel.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/graph_parallel.png ## /.github/static/img/eino/lark_group_en.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/lark_group_en.png ## /.github/static/img/eino/lark_group_zh.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/lark_group_zh.png ## /.github/static/img/eino/react.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/react.png ## /.github/static/img/eino/simple_chain.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/simple_chain.png ## /.github/static/img/eino/stream.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/stream.png ## /.github/static/img/eino/tool_call_graph.png Binary file available at https://raw.githubusercontent.com/cloudwego/eino/refs/heads/main/.github/static/img/eino/tool_call_graph.png ## /.github/workflows/pr-check.yml ```yml path="/.github/workflows/pr-check.yml" name: Pull Request Check on: [ pull_request ] jobs: compliant: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Check License Header uses: apache/skywalking-eyes/header@v0.4.0 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - name: Check Spell uses: crate-ci/typos@master golangci-lint: runs-on: ubuntu-latest permissions: contents: write pull-requests: write repository-projects: write steps: - uses: actions/checkout@v4 - name: Set up Go uses: actions/setup-go@v5 with: go-version: stable # for self-hosted, the cache path is shared across projects # and it works well without the cache of github actions # Enable it if we're going to use Github only cache: true - name: Golangci Lint # https://golangci-lint.run/ uses: golangci/golangci-lint-action@v6 with: version: latest args: --timeout 5m ``` ## /.github/workflows/tests.yml ```yml path="/.github/workflows/tests.yml" name: EinoTests on: pull_request: push: branches: - main env: DEFAULT_GO_VERSION: "1.18" jobs: unit-test: name: eino-unit-test runs-on: ubuntu-latest permissions: contents: write pull-requests: write repository-projects: write env: COVERAGE_FILE: coverage.out BREAKDOWN_FILE: main.breakdown steps: - uses: actions/checkout@v4 - name: Set up Go uses: actions/setup-go@v5 with: go-version: ${{ env.DEFAULT_GO_VERSION }} - name: Exec Go Test run: | modules=`find . -name "go.mod" -exec dirname {} \;` echo $modules list="" coverpkg="" if [[ ! -f "go.work" ]];then go work init;fi for module in $modules; do go work use $module; list=$module"/... "$list; coverpkg=$module"/...,"$coverpkg; done go work sync go test -race -v -coverprofile=${{ env.COVERAGE_FILE }} -gcflags="all=-l -N" -coverpkg=$coverpkg $list - name: Download Artifact (main.breakdown) id: download-main-breakdown uses: actions/download-artifact@v4 continue-on-error: true with: name: ${{ env.BREAKDOWN_FILE }} - name: Create main.breakdown If Not Exist run: | if [ ! -f ${{ env.BREAKDOWN_FILE }} ]; then echo "${{ env.BREAKDOWN_FILE }} not found. Creating an empty file." touch ${{ env.BREAKDOWN_FILE }} else echo "${{ env.BREAKDOWN_FILE }} found." fi - name: Calculate Coverage id: coverage uses: vladopajic/go-test-coverage@v2 with: config: ./.testcoverage.yml profile: ${{ env.COVERAGE_FILE}} breakdown-file-name: ${{ github.ref_name == 'main' && env.BREAKDOWN_FILE || '' }} diff-base-breakdown-file-name: ${{ env.BREAKDOWN_FILE }} # to generate and embed coverage badges in markdown files git-token: ${{ github.ref_name == 'main' && secrets.GITHUB_TOKEN || '' }} git-branch: badges - name: Upload Artifact (main.breakdown) uses: actions/upload-artifact@v4 if: github.ref_name == 'main' with: name: ${{ env.BREAKDOWN_FILE }} path: ${{ env.BREAKDOWN_FILE }} if-no-files-found: error - name: Find If coverage Report Exist if: ${{ github.event.pull_request.number != null }} uses: peter-evans/find-comment@v3 id: fc with: issue-number: ${{ github.event.pull_request.number }} comment-author: 'github-actions[bot]' body-includes: '📊 Coverage Report' - name: Send Coverage Report if: ${{ github.event.pull_request.number != null }} uses: peter-evans/create-or-update-comment@v4 with: token: ${{ secrets.GITHUB_TOKEN }} issue-number: ${{ github.event.pull_request.number }} comment-id: ${{ steps.fc.outputs.comment-id || '' }} edit-mode: replace body: | ## 📊 Coverage Report: \`\`\` ${{ steps.coverage.outputs.report && fromJSON(steps.coverage.outputs.report) || 'No coverage report available' }} \`\`\` - name: Check Coverage if: steps.coverage.outcome == 'failure' shell: bash run: echo "coverage check failed" && exit 1 benchmark-test: runs-on: ubuntu-latest permissions: contents: write pull-requests: write repository-projects: write steps: - uses: actions/checkout@v4 - name: Set up Go uses: actions/setup-go@v5 with: go-version: ${{ env.DEFAULT_GO_VERSION }} - name: Run Benchmark Tests run: go test -bench=. -benchmem -run=none ./... compatibility-test: strategy: matrix: go: [ "1.19", "1.20", "1.21", "1.22" ] runs-on: ubuntu-latest permissions: contents: write pull-requests: write repository-projects: write steps: - uses: actions/checkout@v4 - name: Set up Go uses: actions/setup-go@v5 with: go-version: ${{ matrix.go }} cache: true - name: Compatibility Test run: | # just basic unit test, no coverage report go test -race ./... api-compatibility: name: api-compatibility-check runs-on: ubuntu-latest permissions: contents: write pull-requests: write repository-projects: write if: github.event_name == 'pull_request' steps: - uses: actions/checkout@v4 with: fetch-depth: 0 - name: Set up Go uses: actions/setup-go@v5 with: go-version: "1.22" - name: Install go-apidiff run: go install github.com/joelanford/go-apidiff@v0.8.2 - name: Check API compatibility id: apidiff run: | BASE_SHA=${{ github.event.pull_request.base.sha }} HEAD_SHA=${{ github.event.pull_request.head.sha }} echo "Checking API compatibility between $BASE_SHA and $HEAD_SHA" go mod tidy if ! DIFF_OUTPUT=$(go-apidiff $BASE_SHA $HEAD_SHA 2>&1); then echo "go-apidiff output: $DIFF_OUTPUT" fi echo "diff_output<> $GITHUB_ENV echo "$DIFF_OUTPUT" >> $GITHUB_ENV echo "EOF" >> $GITHUB_ENV if echo "$DIFF_OUTPUT" | grep -q "Incompatible changes:"; then echo "has_breaking_changes=true" >> $GITHUB_OUTPUT else echo "has_breaking_changes=false" >> $GITHUB_OUTPUT fi - name: Create Review Thread if: steps.apidiff.outputs.has_breaking_changes == 'true' continue-on-error: true uses: actions/github-script@v7 with: github-token: ${{ secrets.GITHUB_TOKEN }} script: | const reviewComments = await github.rest.pulls.listReviewComments({ owner: context.repo.owner, repo: context.repo.repo, pull_number: context.issue.number }); const existingPackageComments = new Map(); for (const comment of reviewComments.data) { if (comment.body.includes('Breaking API Changes Detected')) { const packageMatch = comment.body.match(/Package: `([^`]+)`/); if (packageMatch) { const pkg = packageMatch[1]; if (!existingPackageComments.has(pkg)) { existingPackageComments.set(pkg, new Set()); } existingPackageComments.get(pkg).add(comment.path); } } } const files = await github.rest.pulls.listFiles({ owner: context.repo.owner, repo: context.repo.repo, pull_number: context.issue.number }); const diffOutput = process.env.diff_output || ''; const breakingChanges = new Map(); let currentPackage = ''; let isInIncompatibleSection = false; const lines = diffOutput.split('\n'); for (let i = 0; i < lines.length; i++) { const line = lines[i].trim(); if (line.startsWith('github.com/')) { currentPackage = line; if (!breakingChanges.has(currentPackage)) { breakingChanges.set(currentPackage, []); } continue; } if (line === 'Incompatible changes:') { isInIncompatibleSection = true; continue; } if (line === '') { isInIncompatibleSection = false; continue; } if (isInIncompatibleSection && line.startsWith('- ')) { const change = line.substring(2); if (currentPackage) { breakingChanges.get(currentPackage).push(change); } } } const changedFiles = files.data; for (const [pkg, changes] of breakingChanges) { if (changes.length === 0) continue; const pkgPath = pkg.split('/').slice(3).join('/'); const matchingFile = changedFiles.find(file => file.filename.includes(pkgPath) ) || changedFiles[0]; const hasCommentForPackage = existingPackageComments.has(pkg) && existingPackageComments.get(pkg).has(matchingFile.filename); if (matchingFile && !hasCommentForPackage) { const changesList = changes.map(change => { const [name, desc] = change.split(':').map(s => s.trim()); return `- **${name}:** ${desc}`; }).join('\n'); const commentBody = [ '🚨 **Breaking API Changes Detected**', '', `Package: \`${pkg}\``, '', 'Incompatible changes:', changesList, '', '
', 'Review Guidelines', '', 'Please ensure that:', '- The changes are absolutely necessary', '- They are properly documented', '- Migration guides are provided if needed', '
', '', '⚠️ Please resolve this thread after reviewing the breaking changes.' ].join('\n'); await github.rest.pulls.createReview({ owner: context.repo.owner, repo: context.repo.repo, pull_number: context.issue.number, event: 'COMMENT', comments: [{ path: matchingFile.filename, position: matchingFile.patch ? matchingFile.patch.split('\n').findIndex(line => line.startsWith('+')) + 1 : 1, body: commentBody }] }); if (!existingPackageComments.has(pkg)) { existingPackageComments.set(pkg, new Set()); } existingPackageComments.get(pkg).add(matchingFile.filename); } } ``` ## /.gitignore ```gitignore path="/.gitignore" # Binaries for programs and plugins *.exe *.exe~ *.dll *.so *.dylib # Test binary, built with `go test -c` *.test # Output of the go coverage tool, specifically when used with LiteIDE *.out # Dependency directories (remove the comment below to include it) # vendor/ # Go workspace file go.work go.work.sum # env file .env # the result of the go build output* output/* # Files generated by IDEs .idea/ *.iml # Vim swap files *.swp # Vscode files .vscode /patches /vendor ``` ## /.golangci.yaml ```yaml path="/.golangci.yaml" # output configuration options output: # Format: colored-line-number|line-number|json|tab|checkstyle|code-climate|junit-xml|github-actions formats: - format: colored-line-number # All available settings of specific linters. # Refer to https://golangci-lint.run/usage/linters linters-settings: gofumpt: # Choose whether to use the extra rules. # Default: false extra-rules: true govet: # Disable analyzers by name. # Run `go tool vet help` to see all analyzers. disable: - stdmethods linters: enable: - goimports - gofmt disable: - errcheck - typecheck - staticcheck - unused - gosimple - ineffassign - gofumpt issues: # include `vendor` `third_party` `testdata` `examples` `Godeps` `builtin` exclude-use-default: true exclude-files: - ".*\\.mock\\.go$" # exclude-dirs: ``` ## /.licenserc.yaml ```yaml path="/.licenserc.yaml" header: license: spdx-id: Apache-2.0 copyright-owner: CloudWeGo Authors template: | /* * Copyright {{ .Year }} CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ paths: - '**/*.go' - '**/*.s' comment: on-failure ``` ## /.testcoverage.yml ```yml path="/.testcoverage.yml" # (optional; but recommended to set) # When specified reported file paths will not contain local prefix in the output. local-prefix: "github.com/cloudwego/eino" # Holds coverage thresholds percentages, values should be in range [0-100]. threshold: # (optional; default 0) # Minimum overall project coverage percentage required. total: 83 package: 30 # (optional; default 0) # Minimum coverage percentage required for individual files. file: 20 by-package: threshold: 30 show-all: false top-n: 5 bottom-n: 5 by-file: threshold: 20 show-all: false top-n: 5 bottom-n: 5 diff: threshold: 80 show-all: true new-code: true modified-code: true # Holds regexp rules which will exclude matched files or packages # from coverage statistics. exclude: paths: - "tests" - "examples/" - "mock/" - "callbacks/interface.go" - "utils/safe" ``` ## /CODE_OF_CONDUCT.md # Contributor Covenant Code of Conduct ## Our Pledge We as members, contributors, and leaders pledge to make participation in our community a harassment-free experience for everyone, regardless of age, body size, visible or invisible disability, ethnicity, sex characteristics, gender identity and expression, level of experience, education, socio-economic status, nationality, personal appearance, race, religion, or sexual identity and orientation. We pledge to act and interact in ways that contribute to an open, welcoming, diverse, inclusive, and healthy community. ## Our Standards Examples of behavior that contributes to a positive environment for our community include: * Demonstrating empathy and kindness toward other people * Being respectful of differing opinions, viewpoints, and experiences * Giving and gracefully accepting constructive feedback * Accepting responsibility and apologizing to those affected by our mistakes, and learning from the experience * Focusing on what is best not just for us as individuals, but for the overall community Examples of unacceptable behavior include: * The use of sexualized language or imagery, and sexual attention or advances of any kind * Trolling, insulting or derogatory comments, and personal or political attacks * Public or private harassment * Publishing others' private information, such as a physical or email address, without their explicit permission * Other conduct which could reasonably be considered inappropriate in a professional setting ## Enforcement Responsibilities Community leaders are responsible for clarifying and enforcing our standards of acceptable behavior and will take appropriate and fair corrective action in response to any behavior that they deem inappropriate, threatening, offensive, or harmful. Community leaders have the right and responsibility to remove, edit, or reject comments, commits, code, wiki edits, issues, and other contributions that are not aligned to this Code of Conduct, and will communicate reasons for moderation decisions when appropriate. ## Scope This Code of Conduct applies within all community spaces, and also applies when an individual is officially representing the community in public spaces. Examples of representing our community include using an official e-mail address, posting via an official social media account, or acting as an appointed representative at an online or offline event. ## Enforcement Instances of abusive, harassing, or otherwise unacceptable behavior may be reported to the community leaders responsible for enforcement at conduct@cloudwego.io. All complaints will be reviewed and investigated promptly and fairly. All community leaders are obligated to respect the privacy and security of the reporter of any incident. ## Enforcement Guidelines Community leaders will follow these Community Impact Guidelines in determining the consequences for any action they deem in violation of this Code of Conduct: ### 1. Correction **Community Impact**: Use of inappropriate language or other behavior deemed unprofessional or unwelcome in the community. **Consequence**: A private, written warning from community leaders, providing clarity around the nature of the violation and an explanation of why the behavior was inappropriate. A public apology may be requested. ### 2. Warning **Community Impact**: A violation through a single incident or series of actions. **Consequence**: A warning with consequences for continued behavior. No interaction with the people involved, including unsolicited interaction with those enforcing the Code of Conduct, for a specified period of time. This includes avoiding interactions in community spaces as well as external channels like social media. Violating these terms may lead to a temporary or permanent ban. ### 3. Temporary Ban **Community Impact**: A serious violation of community standards, including sustained inappropriate behavior. **Consequence**: A temporary ban from any sort of interaction or public communication with the community for a specified period of time. No public or private interaction with the people involved, including unsolicited interaction with those enforcing the Code of Conduct, is allowed during this period. Violating these terms may lead to a permanent ban. ### 4. Permanent Ban **Community Impact**: Demonstrating a pattern of violation of community standards, including sustained inappropriate behavior, harassment of an individual, or aggression toward or disparagement of classes of individuals. **Consequence**: A permanent ban from any sort of public interaction within the community. ## Attribution This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 2.0, available at https://www.contributor-covenant.org/version/2/0/code_of_conduct.html. Community Impact Guidelines were inspired by [Mozilla's code of conduct enforcement ladder](https://github.com/mozilla/diversity). [homepage]: https://www.contributor-covenant.org For answers to common questions about this code of conduct, see the FAQ at https://www.contributor-covenant.org/faq. Translations are available at https://www.contributor-covenant.org/translations. ## /CONTRIBUTING.md # How to Contribute ## Your First Pull Request We use GitHub for our codebase. You can start by reading [How To Pull Request](https://docs.github.com/en/github/collaborating-with-issues-and-pull-requests/about-pull-requests). ## Branch Organization We use [git-flow](https://nvie.com/posts/a-successful-git-branching-model/) as our branch organization, as known as [FDD](https://en.wikipedia.org/wiki/Feature-driven_development) ## Bugs ### 1. How to Find Known Issues We are using [Github Issues](https://github.com/cloudwego/{project_name}/issues) for our public bugs. We keep a close eye on this and try to make it clear when we have an internal fix in progress. Before filing a new task, try to make sure your problem doesn’t already exist. ### 2. Reporting New Issues Providing a reduced test code is a recommended way for reporting issues. Then can place in: - Just in issues - [Golang Playground](https://play.golang.org/) ### 3. Security Bugs Please do not report the safe disclosure of bugs to public issues. Contact us by [Support Email](mailto:conduct@cloudwego.io) ## How to Get in Touch - [Email](mailto:conduct@cloudwego.io) ## Submit a Pull Request Before you submit your Pull Request (PR) consider the following guidelines: 1. Search [GitHub](https://github.com/cloudwego/{project_name}/pulls) for an open or closed PR that relates to your submission. You don't want to duplicate existing efforts. 2. Be sure that an issue describes the problem you're fixing, or documents the design for the feature you'd like to add. Discussing the design upfront helps to ensure that we're ready to accept your work. 3. [Fork](https://docs.github.com/en/github/getting-started-with-github/fork-a-repo) the cloudwego {project_name} repo. 4. In your forked repository, make your changes in a new git branch: ``` git checkout -b my-fix-branch develop ``` 5. Create your patch, including appropriate test cases. 6. Follow our [Style Guides](#code-style-guides). 7. Commit your changes using a descriptive commit message that follows [AngularJS Git Commit Message Conventions](https://docs.google.com/document/d/1QrDFcIiPjSLDn3EL15IJygNPiHORgU1_OOAqWjiDU5Y/edit). Adherence to these conventions is necessary because release notes are automatically generated from these messages. 8. Push your branch to GitHub: ``` git push origin my-fix-branch ``` 9. In GitHub, send a pull request to `{project_name}:develop` ## Contribution Prerequisites - Our development environment keeps up with [Go Official](https://golang.org/project/). - You need fully checking with lint tools before submit your pull request. [gofmt](https://golang.org/pkg/cmd/gofmt/) and [golangci-lint](https://github.com/golangci/golangci-lint) - You are familiar with [GitHub](https://github.com) - Maybe you need familiar with [Actions](https://github.com/features/actions)(our default workflow tool). ## Code Style Guides - [Effective Go](https://golang.org/doc/effective_go) - [Go Code Review Comments](https://github.com/golang/go/wiki/CodeReviewComments) ## /LICENSE-APACHE ``` path="/LICENSE-APACHE" Apache License Version 2.0, January 2004 http://www.apache.org/licenses/ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION 1. Definitions. "License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document. "Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License. "Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity. "You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License. "Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files. "Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types. "Work" shall mean the work of authorship, whether in Source or Object form, made available under the License, as indicated by a copyright notice that is included in or attached to the work (an example is provided in the Appendix below). "Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. For the purposes of this License, Derivative Works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work and Derivative Works thereof. "Contribution" shall mean any work of authorship, including the original version of the Work and any modifications or additions to that Work or Derivative Works thereof, that is intentionally submitted to Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submitted" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or otherwise designated in writing by the copyright owner as "Not a Contribution." "Contributor" shall mean Licensor and any individual or Legal Entity on behalf of whom a Contribution has been received by Licensor and subsequently incorporated within the Work. 2. Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form. 3. Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work, where such license applies only to those patent claims licensable by such Contributor that are necessarily infringed by their Contribution(s) alone or by combination of their Contribution(s) with the Work to which such Contribution(s) was submitted. If You institute patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Work or a Contribution incorporated within the Work constitutes direct or contributory patent infringement, then any patent licenses granted to You under this License for that Work shall terminate as of the date such litigation is filed. 4. Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions: (a) You must give any other recipients of the Work or Derivative Works a copy of this License; and (b) You must cause any modified files to carry prominent notices stating that You changed the files; and (c) You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work, excluding those notices that do not pertain to any part of the Derivative Works; and (d) If the Work includes a "NOTICE" text file as part of its distribution, then any Derivative Works that You distribute must include a readable copy of the attribution notices contained within such NOTICE file, excluding those notices that do not pertain to any part of the Derivative Works, in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. The contents of the NOTICE file are for informational purposes only and do not modify the License. You may add Your own attribution notices within Derivative Works that You distribute, alongside or as an addendum to the NOTICE text from the Work, provided that such additional attribution notices cannot be construed as modifying the License. You may add Your own copyright statement to Your modifications and may provide additional or different license terms and conditions for use, reproduction, or distribution of Your modifications, or for any such Derivative Works as a whole, provided Your use, reproduction, and distribution of the Work otherwise complies with the conditions stated in this License. 5. Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. Notwithstanding the above, nothing herein shall supersede or modify the terms of any separate license agreement you may have executed with Licensor regarding such Contributions. 6. Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work and reproducing the content of the NOTICE file. 7. Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied, including, without limitation, any warranties or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are solely responsible for determining the appropriateness of using or redistributing the Work and assume any risks associated with Your exercise of permissions under this License. 8. Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or consequential damages of any character arising as a result of this License or out of the use or inability to use the Work (including but not limited to damages for loss of goodwill, work stoppage, computer failure or malfunction, or any and all other commercial damages or losses), even if such Contributor has been advised of the possibility of such damages. 9. Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. However, in accepting such obligations, You may act only on Your own behalf and on Your sole responsibility, not on behalf of any other Contributor, and only if You agree to indemnify, defend, and hold each Contributor harmless for any liability incurred by, or claims asserted against, such Contributor by reason of your accepting any such warranty or additional liability. END OF TERMS AND CONDITIONS ``` ## /README.md # Eino ![coverage](https://raw.githubusercontent.com/cloudwego/eino/badges/.badges/main/coverage.svg) [![Release](https://img.shields.io/github/v/release/cloudwego/eino)](https://github.com/cloudwego/eino/releases) [![WebSite](https://img.shields.io/website?up_message=cloudwego&url=https%3A%2F%2Fwww.cloudwego.io%2F)](https://www.cloudwego.io/) [![License](https://img.shields.io/github/license/cloudwego/eino)](https://github.com/cloudwego/eino/blob/main/LICENSE) [![Go Report Card](https://goreportcard.com/badge/github.com/cloudwego/eino)](https://goreportcard.com/report/github.com/cloudwego/eino) [![OpenIssue](https://img.shields.io/github/issues/cloudwego/eino)](https://github.com/cloudwego/kitex/eino) [![ClosedIssue](https://img.shields.io/github/issues-closed/cloudwego/eino)](https://github.com/cloudwego/eino/issues?q=is%3Aissue+is%3Aclosed) ![Stars](https://img.shields.io/github/stars/cloudwego/eino) ![Forks](https://img.shields.io/github/forks/cloudwego/eino) English | [中文](README.zh_CN.md) # Overview **Eino['aino]** (pronounced similarly to "I know") aims to be the ultimate LLM application development framework in Golang. Drawing inspirations from many excellent LLM application development frameworks in the open-source community such as LangChain & LlamaIndex, etc., as well as learning from cutting-edge research and real world applications, Eino offers an LLM application development framework that emphasizes on simplicity, scalability, reliability and effectiveness that better aligns with Golang programming conventions. What Eino provides are: - a carefully curated list of **component** abstractions and implementations that can be easily reused and combined to build LLM applications - a powerful **composition** framework that does the heavy lifting of strong type checking, stream processing, concurrency management, aspect injection, option assignment, etc. for the user. - a set of meticulously designed **API** that obsesses on simplicity and clarity. - an ever-growing collection of best practices in the form of bundled **flows** and **examples**. - a useful set of tools that covers the entire development cycle, from visualized development and debugging to online tracing and evaluation. With the above arsenal, Eino can standardize, simplify, and improve efficiency at different stages of the AI application development cycle: ![](.github/static/img/eino/eino_concept.jpeg) # A quick walkthrough Use a component directly: ```Go model, _ := openai.NewChatModel(ctx, config) // create an invokable LLM instance message, _ := model.Generate(ctx, []*Message{ SystemMessage("you are a helpful assistant."), UserMessage("what does the future AI App look like?")}) ``` Of course, you can do that, Eino provides lots of useful components to use out of the box. But you can do more by using orchestration, for three reasons: - orchestration encapsulates common patterns of LLM application. - orchestration solves the difficult problem of processing stream response by the LLM. - orchestration handles type safety, concurrency management, aspect injection and option assignment for you. Eino provides two set of APIs for orchestration | API | Characteristics and usage | | -------- |-----------------------------------------------------------------------| | Chain | Simple chained directed graph that can only go forward. | | Graph | Cyclic or Acyclic directed graph. Powerful and flexible. | Let's create a simple chain: a ChatTemplate followed by a ChatModel. ![](.github/static/img/eino/simple_chain.png) ```Go chain, _ := NewChain[map[string]any, *Message](). AppendChatTemplate(prompt). AppendChatModel(model). Compile(ctx) chain.Invoke(ctx, map[string]any{"query": "what's your name?"}) ``` Now let's create a graph that uses a ChatModel to generate answer or tool calls, then uses a ToolsNode to execute those tools if needed. ![](.github/static/img/eino/tool_call_graph.png) ```Go graph := NewGraph[map[string]any, *schema.Message]() _ = graph.AddChatTemplateNode("node_template", chatTpl) _ = graph.AddChatModelNode("node_model", chatModel) _ = graph.AddToolsNode("node_tools", toolsNode) _ = graph.AddLambdaNode("node_converter", takeOne) _ = graph.AddEdge(START, "node_template") _ = graph.AddEdge("node_template", "node_model") _ = graph.AddBranch("node_model", branch) _ = graph.AddEdge("node_tools", "node_converter") _ = graph.AddEdge("node_converter", END) compiledGraph, err := graph.Compile(ctx) if err != nil { return err } out, err := r.Invoke(ctx, map[string]any{"query":"Beijing's weather this weekend"}) ``` Now let's create a 'ReAct' agent: A ChatModel binds to Tools. It receives input Messages and decides independently whether to call the Tool or output the final result. The execution result of the Tool will again become the input Message for the ChatModel and serve as the context for the next round of independent judgment. ![](.github/static/img/eino/react.png) We provide a complete implementation for ReAct Agent out of the box in the `flow` package. Check out the code here: [flow/agent/react](https://github.com/cloudwego/eino/blob/main/flow/agent/react/react.go) Our implementation of ReAct Agent uses Eino's **graph orchestration** exclusively, which provides the following benefits out of the box: - Type checking: it makes sure the two nodes' input and output types match at compile time. - Stream processing: concatenates message stream before passing to chatModel and toolsNode if needed, and copies the stream into callback handlers. - Concurrency management: the shared state can be safely read and written because the StatePreHandler is concurrency safe. - Aspect injection: injects callback aspects before and after the execution of ChatModel if the specified ChatModel implementation hasn't injected itself. - Option assignment: call options are assigned either globally, to specific component type or to specific node. For example, you could easily extend the compiled graph with callbacks: ```Go handler := NewHandlerBuilder(). OnStartFn( func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context) { log.Infof("onStart, runInfo: %v, input: %v", info, input) }). OnEndFn( func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context) { log.Infof("onEnd, runInfo: %v, out: %v", info, output) }). Build() compiledGraph.Invoke(ctx, input, WithCallbacks(handler)) ``` or you could easily assign options to different nodes: ```Go // assign to All nodes compiledGraph.Invoke(ctx, input, WithCallbacks(handler)) // assign only to ChatModel nodes compiledGraph.Invoke(ctx, input, WithChatModelOption(WithTemperature(0.5)) // assign only to node_1 compiledGraph.Invoke(ctx, input, WithCallbacks(handler).DesignateNode("node_1")) ``` # Key Features ## Rich Components - Encapsulates common building blocks into **component abstractions**, each have multiple **component implementations** that are ready to be used out of the box. - component abstractions such as ChatModel, Tool, ChatTemplate, Retriever, Document Loader, Lambda, etc. - each component type has an interface of its own: defined Input & Output Type, defined Option type, and streaming paradigms that make sense. - implementations are transparent. Abstractions are all you care about when orchestrating components together. - Implementations can be nested and captures complex business logic. - ReAct Agent, MultiQueryRetriever, Host MultiAgent, etc. They consist of multiple components and non-trivial business logic. - They are still transparent from the outside. A MultiQueryRetriever can be used anywhere that accepts a Retriever. ## Powerful Orchestration - Data flows from Retriever / Document Loaders / ChatTemplate to ChatModel, then flows to Tools and parsed as Final Answer. This directed, controlled flow of data through multiple components can be implemented through **graph orchestration**. - Component instances are graph nodes, and edges are data flow channels. - Graph orchestration is powerful and flexible enough to implement complex business logic: - type checking, stream processing, concurrency management, aspect injection and option assignment are handled by the framework. - branch out execution at runtime, read and write global state, or do field level data mapping using workflow(currently in alpha stage). ## Complete Stream Processing - Stream processing is important because ChatModel outputs chunks of messages in real time as it generates them. It's especially important with orchestration because more components need to handle streaming data. - Eino automatically **concatenates** stream chunks for downstream nodes that only accepts non-stream input, such as ToolsNode. - Eino automatically **boxes** non stream into stream when stream is needed during graph execution. - Eino automatically **merges** multiple streams as they converge into a single downward node. - Eino automatically **copies** stream as they fan out to different downward node, or is passed to callback handlers. - Orchestration elements such as **branch** and **state handlers** are also stream aware. - With these streaming processing abilities, the streaming paradigms of components themselves become transparent to the user. - A compiled Graph can run with 4 different streaming paradigms: | Streaming Paradigm | Explanation | | ------------------ | --------------------------------------------------------------------------- | | Invoke | Accepts non-stream type I and returns non-stream type O | | Stream | Accepts non-stream type I and returns stream type StreamReader[O] | | Collect | Accepts stream type StreamReader[I] and returns non-stream type O | | Transform | Accepts stream type StreamReader[I] and returns stream type StreamReader[O] | ## Highly Extensible Aspects (Callbacks) - Aspects handle cross-cutting concerns such as logging, tracing, metrics, etc., as well as exposing internal details of component implementations. - Five aspects are supported: **OnStart, OnEnd, OnError, OnStartWithStreamInput, OnEndWithStreamOutput**. - Developers can easily create custom callback handlers, add them during graph run via options, and they will be invoked during graph run. - Graph can also inject aspects to those component implementations that do not support callbacks on their own. # Eino Framework Structure ![](.github/static/img/eino/eino_framework.jpeg) The Eino framework consists of several parts: - Eino(this repo): Contains Eino's type definitions, streaming mechanism, component abstractions, orchestration capabilities, aspect mechanisms, etc. - [EinoExt](https://github.com/cloudwego/eino-ext): Component implementations, callback handlers implementations, component usage examples, and various tools such as evaluators, prompt optimizers. - [Eino Devops](https://github.com/cloudwego/eino-ext/tree/main/devops): visualized developing, visualized debugging etc. - [EinoExamples](https://github.com/cloudwego/eino-examples) is the repo containing example applications and best practices for Eino. ## Detailed Documentation For learning and using Eino, we provide a comprehensive Eino User Manual to help you quickly understand the concepts in Eino and master the skills of developing AI applications based on Eino. Start exploring through the [Eino User Manual](https://www.cloudwego.io/zh/docs/eino/) now! For a quick introduction to building AI applications with Eino, we recommend starting with [Eino: Quick Start](https://www.cloudwego.io/zh/docs/eino/quick_start/) ## Dependencies - Go 1.18 and above. - Eino relies on [kin-openapi](https://github.com/getkin/kin-openapi) 's OpenAPI JSONSchema implementation. In order to remain compatible with Go 1.18, we have fixed kin-openapi's version to be v0.118.0. ## Security If you discover a potential security issue in this project, or think you may have discovered a security issue, we ask that you notify Bytedance Security via our [security center](https://security.bytedance.com/src) or [vulnerability reporting email](sec@bytedance.com). Please do **not** create a public GitHub issue. ## Contact US - How to become a member: [COMMUNITY MEMBERSHIP](https://github.com/cloudwego/community/blob/main/COMMUNITY_MEMBERSHIP.md) - Issues: [Issues](https://github.com/cloudwego/eino/issues) - Lark: Scan the QR code below with [Register Feishu](https://www.feishu.cn/en/) to join our CloudWeGo/eino user group.     LarkGroup ## License This project is licensed under the [Apache-2.0 License](LICENSE-APACHE). ## /README.zh_CN.md # Eino ![coverage](https://raw.githubusercontent.com/cloudwego/eino/badges/.badges/main/coverage.svg) [![Release](https://img.shields.io/github/v/release/cloudwego/eino)](https://github.com/cloudwego/eino/releases) [![WebSite](https://img.shields.io/website?up_message=cloudwego&url=https%3A%2F%2Fwww.cloudwego.io%2F)](https://www.cloudwego.io/) [![License](https://img.shields.io/github/license/cloudwego/eino)](https://github.com/cloudwego/eino/blob/main/LICENSE) [![Go Report Card](https://goreportcard.com/badge/github.com/cloudwego/eino)](https://goreportcard.com/report/github.com/cloudwego/eino) [![OpenIssue](https://img.shields.io/github/issues/cloudwego/eino)](https://github.com/cloudwego/kitex/eino) [![ClosedIssue](https://img.shields.io/github/issues-closed/cloudwego/eino)](https://github.com/cloudwego/eino/issues?q=is%3Aissue+is%3Aclosed) ![Stars](https://img.shields.io/github/stars/cloudwego/eino) ![Forks](https://img.shields.io/github/forks/cloudwego/eino) [English](README.md) | 中文 # 简介 **Eino['aino]**（谐音 “I know”）旨在成为用 Go 语言编写的终极大型语言模型（LLM）应用开发框架。它从开源社区中的诸多优秀 LLM 应用开发框架，如 LangChain 和 LlamaIndex 等获取灵感，同时借鉴前沿研究成果与实际应用，提供了一个强调简洁性、可扩展性、可靠性与有效性，且更符合 Go 语言编程惯例的 LLM 应用开发框架。 Eino 提供的价值如下： - 精心整理的一系列 **组件（component）** 抽象与实现，可轻松复用与组合，用于构建 LLM 应用。 - 强大的 **编排（orchestration）** 框架，为用户承担繁重的类型检查、流式处理、并发管理、切面注入、选项赋值等工作。 - 一套精心设计、注重简洁明了的 **API**。 - 以集成 **流程（flow）** 和 **示例（example）** 形式不断扩充的最佳实践集合。 - 一套实用 **工具（DevOps tools）**，涵盖从可视化开发与调试到在线追踪与评估的整个开发生命周期。 借助上述能力和工具，Eino 能够在人工智能应用开发生命周期的不同阶段实现标准化、简化操作并提高效率： ![](.github/static/img/eino/eino_concept.jpeg) # 快速上手 直接使用组件： ```Go model, _ := openai.NewChatModel(ctx, config) // create an invokable LLM instance message, _ := model.Generate(ctx, []*Message{ SystemMessage("you are a helpful assistant."), UserMessage("what does the future AI App look like?")}) ``` 当然，你可以这样用，Eino 提供了许多开箱即用的有用组件。但通过使用编排功能，你能实现更多，原因有三： - 编排封装了大语言模型（LLM）应用的常见模式。 - 编排解决了处理大语言模型流式响应这一难题。 - 编排为你处理类型安全、并发管理、切面注入以及选项赋值等问题。 Eino 提供了两组用于编排的 API： | API | 特性和使用场景 | | -------- |-----------------------------| | Chain | 简单的链式有向图，只能向前推进。 | | Graph | 循环或非循环有向图。功能强大且灵活。 | 我们来创建一个简单的 chain: 一个模版（ChatTemplate）接一个大模型（ChatModel）。 ![](.github/static/img/eino/simple_chain.png) ```Go chain, _ := NewChain[map[string]any, *Message](). AppendChatTemplate(prompt). AppendChatModel(model). Compile(ctx) chain.Invoke(ctx, map[string]any{"query": "what's your name?"}) ``` 现在，我们来创建一个 Graph，先用一个 ChatModel 生成回复或者 Tool 调用指令，如生成了 Tool 调用指令，就用一个 ToolsNode 执行这些 Tool。 ![](.github/static/img/eino/tool_call_graph.png) ```Go graph := NewGraph[map[string]any, *schema.Message]() _ = graph.AddChatTemplateNode("node_template", chatTpl) _ = graph.AddChatModelNode("node_model", chatModel) _ = graph.AddToolsNode("node_tools", toolsNode) _ = graph.AddLambdaNode("node_converter", takeOne) _ = graph.AddEdge(START, "node_template") _ = graph.AddEdge("node_template", "node_model") _ = graph.AddBranch("node_model", branch) _ = graph.AddEdge("node_tools", "node_converter") _ = graph.AddEdge("node_converter", END) compiledGraph, err := graph.Compile(ctx) if err != nil { return err } out, err := r.Invoke(ctx, map[string]any{"query":"Beijing's weather this weekend"}) ``` 现在，咱们来创建一个 “ReAct” 智能体：一个 ChatModel 绑定了一些 Tool。它接收输入的消息，自主判断是调用 Tool 还是输出最终结果。Tool 的执行结果会再次成为聊天模型的输入消息，并作为下一轮自主判断的上下文。 ![](.github/static/img/eino/react.png) 我们在 Eino 的 `flow` 包中提供了开箱即用的 ReAct 智能体的完整实现。代码参见： [flow/agent/react](https://github.com/cloudwego/eino/blob/main/flow/agent/react/react.go) 我们的 ReAct 智能体实现完全基于 Eino 的编排能力。通过使用 Eino 编排，我们可以自动获得如下能力: - **类型检查**：在编译时确保两个节点的输入和输出类型匹配。 - **流处理**：如有需要，在将消息流传递给 ChatModel 和 ToolsNode 节点之前进行拼接，以及将该流复制到callback handler 中。 - **并发管理**：由于 StatePreHandler是线程安全的，共享的 state 可以被安全地读写。 - **切面注入**：如果指定的 ChatModel 实现未自行注入，会在 ChatModel 执行之前和之后注入回调切面。 - **选项赋值**：调用 Option 可以全局设置，也可以针对特定组件类型或特定节点进行设置。 例如，你可以轻松地通过回调扩展已编译的图： ```Go handler := NewHandlerBuilder(). OnStartFn( func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context) { log.Infof("onStart, runInfo: %v, input: %v", info, input) }). OnEndFn( func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context) { log.Infof("onEnd, runInfo: %v, out: %v", info, output) }). Build() compiledGraph.Invoke(ctx, input, WithCallbacks(handler)) ``` 或者你可以轻松地为不同节点分配选项： ```Go // assign to All nodes compiledGraph.Invoke(ctx, input, WithCallbacks(handler)) // assign only to ChatModel nodes compiledGraph.Invoke(ctx, input, WithChatModelOption(WithTemperature(0.5)) // assign only to node_1 compiledGraph.Invoke(ctx, input, WithCallbacks(handler).DesignateNode("node_1")) ``` # 关键特性 ## 丰富的组件 - 将常见的构建模块封装为**组件抽象**，每个组件抽象都有多个可开箱即用的**组件实现**。 - 诸如 ChatModel、Tool、ChatTemplate、Retriever、Document Loader、Lambda 等组件抽象。 - 每种组件类型都有其自身的接口：定义了输入和输出类型、定义了选项类型，以及合理的流处理范式。 - 实现细节是透明的。在编排组件时，你只需关注抽象层面。 - 实现可以嵌套，并包含复杂的业务逻辑。 - ReAct Agent、MultiQueryRetriever、Host MultiAgent 等。它们由多个组件和复杂的业务逻辑构成。 - 从外部看，它们的实现细节依然透明。例如在任何接受 Retriever 的地方，都可以使用 MultiQueryRetriever。 ## 强大的编排 (Graph/Chain/Workflow) - 数据从 Retriever / Document Loader / ChatTemplate 流向 ChatModel，接着流向 Tool ，并被解析为最终答案。这种通过多个组件的有向、可控的数据流，可以通过**图编排**来实现。 - 组件实例是图的**节点（Node）**，而**边（Edge）**则是数据流通道。 - 图编排功能强大且足够灵活，能够实现复杂的业务逻辑： - **类型检查、流处理、并发管理、切面注入和选项分配**都由框架处理。 - 在运行时进行**分支（Branch）**执行、读写全局**状态（State）**，或者使用工作流进行字段级别的数据映射。 ## 完整的流式处理能力 - 流式处理（Stream Processing）很重要，因为 ChatModel 在生成消息时会实时输出消息块。在编排场景下会尤为重要，因为更多的组件需要处理流式数据。 - 对于只接受非流式输入的下游节点（如 ToolsNode），Eino 会自动将流 **拼接（Concatenate）** 起来。 - 在图执行过程中，当需要流时，Eino 会自动将非流式**转换**为流式。 - 当多个流汇聚到一个下游节点时，Eino 会自动 **合并（Merge）** 这些流。 - 当流分散到不同的下游节点或传递给回调处理器时，Eino 会自动 **复制（Copy）** 这些流。 - 如 **分支（Branch）** 、或 **状态处理器（StateHandler）** 等编排元素，也能够感知和处理流。 - 借助上述流式处理能力，组件本身的流式处理范式变的对用户透明。 - 经过编译的 Graph 可以用 4 种不同的流式范式来运行： | 流处理范式 | 解释 | |-----------|-----------------------------------------------| | Invoke | 接收非流类型 I ，返回非流类型 O | | Stream | 接收非流类型 I ， 返回流类型 StreamReader[O] | | Collect | 接收流类型 StreamReader[I] ， 返回非流类型 O | | Transform | 接收流类型 StreamReader[I] ， 返回流类型 StreamReader[O] | ## 易扩展的切面（Callbacks） - 切面用于处理诸如日志记录、追踪、指标统计等横切面关注点，同时也用于暴露组件实现的内部细节。 - 支持五种切面：**OnStart、OnEnd、OnError、OnStartWithStreamInput、OnEndWithStreamOutput**。 - 开发者可以轻松创建自定义回调处理程序，在图运行期间通过 Option 添加它们，这些处理程序会在图运行时被调用。 - 图还能将切面注入到那些自身不支持回调的组件实现中。 # Eino 框架结构 ![](.github/static/img/eino/eino_framework.jpeg) Eino 框架由几个部分组成： - Eino（本代码仓库）：包含类型定义、流处理机制、组件抽象、编排功能、切面机制等。 - [EinoExt](https://github.com/cloudwego/eino-ext)：组件实现、回调处理程序实现、组件使用示例，以及各种工具，如评估器、提示优化器等。 - [Eino Devops](https://github.com/cloudwego/eino-ext/tree/main/devops)：可视化开发、可视化调试等。 - [EinoExamples](https://github.com/cloudwego/eino-examples)：是包含示例应用程序和最佳实践的代码仓库。 ## 详细文档 针对 Eino 的学习和使用，我们提供了完善的 Eino用户手册，帮助大家快速理解 Eino 中的概念，掌握基于 Eino 开发设计 AI 应用的技能，赶快通过[Eino 用户手册](https://www.cloudwego.io/zh/docs/eino/)尝试使用吧~。 若想快速上手，了解 通过 Eino 构建 AI 应用的过程，推荐先阅读[Eino: 快速开始](https://www.cloudwego.io/zh/docs/eino/quick_start/) ## 依赖说明 - Go 1.18 及以上版本 - Eino 依赖了 [kin-openapi](https://github.com/getkin/kin-openapi) 的 OpenAPI JSONSchema 实现。为了能够兼容 Go 1.18 版本，我们将 kin-openapi 的版本固定在了 v0.118.0。 ## 安全 如果你在该项目中发现潜在的安全问题，或你认为可能发现了安全问题，请通过我们的[安全中心](https://security.bytedance.com/src) 或[漏洞报告邮箱](sec@bytedance.com)通知字节跳动安全团队。 请**不要**创建公开的 GitHub Issue。 ## 联系我们 - 如何成为 member: [COMMUNITY MEMBERSHIP](https://github.com/cloudwego/community/blob/main/COMMUNITY_MEMBERSHIP.md) - Issues: [Issues](https://github.com/cloudwego/eino/issues) - 飞书用户群（[注册飞书](https://www.feishu.cn/)后扫码进群）     LarkGroup ## 开源许可证 本项目依据 [Apache-2.0 许可证](LICENSE-APACHE) 授权。 ## /_typos.toml ```toml path="/_typos.toml" # Typo check: https://github.com/crate-ci/typos [default] [default.extend-words] Invokable = "Invokable" invokable = "invokable" InvokableLambda = "InvokableLambda" InvokableRun = "InvokableRun" typ = "typ" [files] extend-exclude = ["go.mod", "go.sum", "check_branch_name.sh"] ``` ## /callbacks/aspect_inject.go ```go path="/callbacks/aspect_inject.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package callbacks import ( "context" "github.com/cloudwego/eino/components" "github.com/cloudwego/eino/internal/callbacks" "github.com/cloudwego/eino/schema" ) // OnStart Fast inject callback input / output aspect for component developer // e.g. // // func (t *testChatModel) Generate(ctx context.Context, input []*schema.Message, opts ...model.Option) (resp *schema.Message, err error) { // defer func() { // if err != nil { // callbacks.OnEnd(ctx, err) // } // }() // // ctx = callbacks.OnStart(ctx, &model.CallbackInput{ // Messages: input, // Tools: nil, // Extra: nil, // }) // // // do smt // // ctx = callbacks.OnEnd(ctx, &model.CallbackOutput{ // Message: resp, // Extra: nil, // }) // // return resp, nil // } // // OnStart invokes the OnStart logic for the particular context, ensuring that all registered // handlers are executed in reverse order (compared to add order) when a process begins. func OnStart[T any](ctx context.Context, input T) context.Context { ctx, _ = callbacks.On(ctx, input, callbacks.OnStartHandle[T], TimingOnStart) return ctx } // OnEnd invokes the OnEnd logic of the particular context, allowing for proper cleanup // and finalization when a process ends. // handlers are executed in normal order (compared to add order). func OnEnd[T any](ctx context.Context, output T) context.Context { ctx, _ = callbacks.On(ctx, output, callbacks.OnEndHandle[T], TimingOnEnd) return ctx } // OnStartWithStreamInput invokes the OnStartWithStreamInput logic of the particular context, ensuring that // every input stream should be closed properly in handler. // handlers are executed in reverse order (compared to add order). func OnStartWithStreamInput[T any](ctx context.Context, input *schema.StreamReader[T]) ( nextCtx context.Context, newStreamReader *schema.StreamReader[T]) { return callbacks.On(ctx, input, callbacks.OnStartWithStreamInputHandle[T], TimingOnStartWithStreamInput) } // OnEndWithStreamOutput invokes the OnEndWithStreamOutput logic of the particular, ensuring that // every input stream should be closed properly in handler. // handlers are executed in normal order (compared to add order). func OnEndWithStreamOutput[T any](ctx context.Context, output *schema.StreamReader[T]) ( nextCtx context.Context, newStreamReader *schema.StreamReader[T]) { return callbacks.On(ctx, output, callbacks.OnEndWithStreamOutputHandle[T], TimingOnEndWithStreamOutput) } // OnError invokes the OnError logic of the particular, notice that error in stream will not represent here. // handlers are executed in normal order (compared to add order). func OnError(ctx context.Context, err error) context.Context { ctx, _ = callbacks.On(ctx, err, callbacks.OnErrorHandle, TimingOnError) return ctx } // EnsureRunInfo ensures the RunInfo in context matches the given type and component. // If the current callback manager doesn't match or doesn't exist, it creates a new one while preserving existing handlers. func EnsureRunInfo(ctx context.Context, typ string, comp components.Component) context.Context { return callbacks.EnsureRunInfo(ctx, typ, comp) } // ReuseHandlers initializes a new context with the provided RunInfo, while using the same handlers already exist. func ReuseHandlers(ctx context.Context, info *RunInfo) context.Context { return callbacks.ReuseHandlers(ctx, info) } // InitCallbacks initializes a new context with the provided RunInfo and handlers. // Any previously set RunInfo and Handlers for this ctx will be overwritten. func InitCallbacks(ctx context.Context, info *RunInfo, handlers ...Handler) context.Context { return callbacks.InitCallbacks(ctx, info, handlers...) } ``` ## /callbacks/aspect_inject_test.go ```go path="/callbacks/aspect_inject_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package callbacks import ( "context" "fmt" "io" "strconv" "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/internal/callbacks" "github.com/cloudwego/eino/schema" ) func TestAspectInject(t *testing.T) { t.Run("ctx without manager", func(t *testing.T) { ctx := context.Background() ctx = OnStart(ctx, 1) ctx = OnEnd(ctx, 2) ctx = OnError(ctx, fmt.Errorf("3")) isr, isw := schema.Pipe[int](2) go func() { for i := 0; i < 10; i++ { isw.Send(i, nil) } isw.Close() }() var nisr *schema.StreamReader[int] ctx, nisr = OnStartWithStreamInput(ctx, isr) j := 0 for { i, err := nisr.Recv() if err == io.EOF { break } assert.NoError(t, err) assert.Equal(t, j, i) j++ } nisr.Close() osr, osw := schema.Pipe[int](2) go func() { for i := 0; i < 10; i++ { osw.Send(i, nil) } osw.Close() }() var nosr *schema.StreamReader[int] ctx, nosr = OnEndWithStreamOutput(ctx, osr) j = 0 for { i, err := nosr.Recv() if err == io.EOF { break } assert.NoError(t, err) assert.Equal(t, j, i) j++ } nosr.Close() }) t.Run("ctx with manager", func(t *testing.T) { ctx := context.Background() cnt := 0 hb := NewHandlerBuilder(). OnStartFn(func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context { cnt += input.(int) return ctx }). OnEndFn(func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context { cnt += output.(int) return ctx }). OnErrorFn(func(ctx context.Context, info *RunInfo, err error) context.Context { v, _ := strconv.ParseInt(err.Error(), 10, 64) cnt += int(v) return ctx }). OnStartWithStreamInputFn(func(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context { for { i, err := input.Recv() if err == io.EOF { break } cnt += i.(int) } input.Close() return ctx }). OnEndWithStreamOutputFn(func(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context { for { o, err := output.Recv() if err == io.EOF { break } cnt += o.(int) } output.Close() return ctx }).Build() ctx = InitCallbacks(ctx, nil, hb) ctx = OnStart(ctx, 1) ctx = OnEnd(ctx, 2) ctx = OnError(ctx, fmt.Errorf("3")) isr, isw := schema.Pipe[int](2) go func() { for i := 0; i < 10; i++ { isw.Send(i, nil) } isw.Close() }() var nisr *schema.StreamReader[int] ctx, nisr = OnStartWithStreamInput(ctx, isr) j := 0 for { i, err := nisr.Recv() if err == io.EOF { break } assert.NoError(t, err) assert.Equal(t, j, i) j++ cnt += i } nisr.Close() osr, osw := schema.Pipe[int](2) go func() { for i := 0; i < 10; i++ { osw.Send(i, nil) } osw.Close() }() var nosr *schema.StreamReader[int] ctx, nosr = OnEndWithStreamOutput(ctx, osr) j = 0 for { i, err := nosr.Recv() if err == io.EOF { break } assert.NoError(t, err) assert.Equal(t, j, i) j++ cnt += i } nosr.Close() assert.Equal(t, 186, cnt) }) } func TestGlobalCallbacksRepeated(t *testing.T) { times := 0 testHandler := NewHandlerBuilder().OnStartFn(func(ctx context.Context, info *callbacks.RunInfo, input callbacks.CallbackInput) context.Context { times++ return ctx }).Build() callbacks.GlobalHandlers = append(callbacks.GlobalHandlers, testHandler) ctx := context.Background() ctx = callbacks.AppendHandlers(ctx, &RunInfo{}) ctx = callbacks.AppendHandlers(ctx, &RunInfo{}) callbacks.On(ctx, "test", callbacks.OnStartHandle[string], TimingOnStart) assert.Equal(t, times, 1) } func TestEnsureRunInfo(t *testing.T) { ctx := context.Background() var name, typ, comp string ctx = InitCallbacks(ctx, &RunInfo{Name: "name", Type: "type", Component: "component"}, NewHandlerBuilder().OnStartFn(func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context { name = info.Name typ = info.Type comp = string(info.Component) return ctx }).Build()) ctx2 := EnsureRunInfo(ctx, "type2", "component2") OnStart(ctx, "") assert.Equal(t, "name", name) assert.Equal(t, "type", typ) assert.Equal(t, "component", comp) OnStart(ctx2, "") assert.Equal(t, "", name) assert.Equal(t, "type2", typ) assert.Equal(t, "component2", comp) } ``` ## /callbacks/doc.go ```go path="/callbacks/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ // Package callbacks provides callback mechanisms for component execution in Eino. // // This package allows you to inject callback handlers at different stages of component execution, // such as start, end, and error handling. It's particularly useful for implementing governance capabilities like logging, monitoring, and metrics collection. // // The package provides two ways to create callback handlers: // // 1. Create a callback handler using HandlerBuilder: // // handler := callbacks.NewHandlerBuilder(). // OnStart(func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context { // // Handle component start // return ctx // }). // OnEnd(func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context { // // Handle component end // return ctx // }). // OnError(func(ctx context.Context, info *RunInfo, err error) context.Context { // // Handle component error // return ctx // }). // OnStartWithStreamInput(func(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context { // // Handle component start with stream input // return ctx // }). // OnEndWithStreamOutput(func(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context { // // Handle component end with stream output // return ctx // }). // Build() // // For this way, you need to convert the callback input types by yourself, and implement the logic for different component types in one handler. // // 2. Use [template.HandlerHelper] to create a handler: // // Package utils/callbacks provides [HandlerHelper] as a convenient way to build callback handlers // for different component types. It allows you to set specific handlers for each component type, // // e.g. // // // Create handlers for specific components // modelHandler := &model.CallbackHandler{ // OnStart: func(ctx context.Context, info *RunInfo, input *model.CallbackInput) context.Context { // log.Printf("Model execution started: %s", info.ComponentName) // return ctx // }, // } // // promptHandler := &prompt.CallbackHandler{ // OnEnd: func(ctx context.Context, info *RunInfo, output *prompt.CallbackOutput) context.Context { // log.Printf("Prompt execution completed: %s", output.Result) // return ctx // }, // } // // // Build the handler using HandlerHelper // handler := callbacks.NewHandlerHelper(). // ChatModel(modelHandler). // Prompt(promptHandler). // Fallback(fallbackHandler). // Handler() // // [HandlerHelper] supports handlers for various component types including: // - Prompt components (via prompt.CallbackHandler) // - Chat model components (via model.CallbackHandler) // - Embedding components (via embedding.CallbackHandler) // - Indexer components (via indexer.CallbackHandler) // - Retriever components (via retriever.CallbackHandler) // - Document loader components (via loader.CallbackHandler) // - Document transformer components (via transformer.CallbackHandler) // - Tool components (via tool.CallbackHandler) // - Graph (via Handler) // - Chain (via Handler) // - Tools node (via Handler) // - Lambda (via Handler) // // Use the handler with a component: // // runnable.Invoke(ctx, input, compose.WithCallbacks(handler)) package callbacks ``` ## /callbacks/handler_builder.go ```go path="/callbacks/handler_builder.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package callbacks import ( "context" "github.com/cloudwego/eino/schema" ) type HandlerBuilder struct { onStartFn func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context onEndFn func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context onErrorFn func(ctx context.Context, info *RunInfo, err error) context.Context onStartWithStreamInputFn func(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context onEndWithStreamOutputFn func(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context } type handlerImpl struct { HandlerBuilder } func (hb *handlerImpl) OnStart(ctx context.Context, info *RunInfo, input CallbackInput) context.Context { return hb.onStartFn(ctx, info, input) } func (hb *handlerImpl) OnEnd(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context { return hb.onEndFn(ctx, info, output) } func (hb *handlerImpl) OnError(ctx context.Context, info *RunInfo, err error) context.Context { return hb.onErrorFn(ctx, info, err) } func (hb *handlerImpl) OnStartWithStreamInput(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context { return hb.onStartWithStreamInputFn(ctx, info, input) } func (hb *handlerImpl) OnEndWithStreamOutput(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context { return hb.onEndWithStreamOutputFn(ctx, info, output) } func (hb *handlerImpl) Needed(_ context.Context, _ *RunInfo, timing CallbackTiming) bool { switch timing { case TimingOnStart: return hb.onStartFn != nil case TimingOnEnd: return hb.onEndFn != nil case TimingOnError: return hb.onErrorFn != nil case TimingOnStartWithStreamInput: return hb.onStartWithStreamInputFn != nil case TimingOnEndWithStreamOutput: return hb.onEndWithStreamOutputFn != nil default: return false } } // NewHandlerBuilder creates and returns a new HandlerBuilder instance. // HandlerBuilder is used to construct a Handler with custom callback functions func NewHandlerBuilder() *HandlerBuilder { return &HandlerBuilder{} } func (hb *HandlerBuilder) OnStartFn( fn func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context) *HandlerBuilder { hb.onStartFn = fn return hb } func (hb *HandlerBuilder) OnEndFn( fn func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context) *HandlerBuilder { hb.onEndFn = fn return hb } func (hb *HandlerBuilder) OnErrorFn( fn func(ctx context.Context, info *RunInfo, err error) context.Context) *HandlerBuilder { hb.onErrorFn = fn return hb } // OnStartWithStreamInputFn sets the callback function to be called. func (hb *HandlerBuilder) OnStartWithStreamInputFn( fn func(ctx context.Context, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context) *HandlerBuilder { hb.onStartWithStreamInputFn = fn return hb } // OnEndWithStreamOutputFn sets the callback function to be called. func (hb *HandlerBuilder) OnEndWithStreamOutputFn( fn func(ctx context.Context, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context) *HandlerBuilder { hb.onEndWithStreamOutputFn = fn return hb } // Build returns a Handler with the functions set in the builder. func (hb *HandlerBuilder) Build() Handler { return &handlerImpl{*hb} } ``` ## /callbacks/interface.go ```go path="/callbacks/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package callbacks import ( "github.com/cloudwego/eino/internal/callbacks" ) // RunInfo contains information about the running component. type RunInfo = callbacks.RunInfo // CallbackInput is the input of the callback. // the type of input is defined by the component. // using type Assert or convert func to convert the input to the right type you want. // e.g. // // CallbackInput in components/model/interface.go is: // type CallbackInput struct { // Messages []*schema.Message // Config *Config // Extra map[string]any // } // // and provide a func of model.ConvCallbackInput() to convert CallbackInput to *model.CallbackInput // in callback handler, you can use the following code to get the input: // // modelCallbackInput := model.ConvCallbackInput(in) // if modelCallbackInput == nil { // // is not a model callback input, just ignore it // return // } type CallbackInput = callbacks.CallbackInput type CallbackOutput = callbacks.CallbackOutput type Handler = callbacks.Handler // InitCallbackHandlers sets the global callback handlers. // It should be called BEFORE any callback handler by user. // It's useful when you want to inject some basic callbacks to all nodes. // Deprecated: Use AppendGlobalHandlers instead. func InitCallbackHandlers(handlers []Handler) { callbacks.GlobalHandlers = handlers } // AppendGlobalHandlers appends the given handlers to the global callback handlers. // This is the preferred way to add global callback handlers as it preserves existing handlers. // The global callback handlers will be executed for all nodes BEFORE user-specific handlers in CallOption. // Note: This function is not thread-safe and should only be called during process initialization. func AppendGlobalHandlers(handlers ...Handler) { callbacks.GlobalHandlers = append(callbacks.GlobalHandlers, handlers...) } // CallbackTiming enumerates all the timing of callback aspects. type CallbackTiming = callbacks.CallbackTiming const ( TimingOnStart CallbackTiming = iota TimingOnEnd TimingOnError TimingOnStartWithStreamInput TimingOnEndWithStreamOutput ) // TimingChecker checks if the handler is needed for the given callback aspect timing. // It's recommended for callback handlers to implement this interface, but not mandatory. // If a callback handler is created by using callbacks.HandlerHelper or handlerBuilder, then this interface is automatically implemented. // Eino's callback mechanism will try to use this interface to determine whether any handlers are needed for the given timing. // Also, the callback handler that is not needed for that timing will be skipped. type TimingChecker = callbacks.TimingChecker ``` ## /callbacks/interface_test.go ```go path="/callbacks/interface_test.go" /* * Copyright 2025 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package callbacks import ( "context" "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/internal/callbacks" ) func TestAppendGlobalHandlers(t *testing.T) { // Clear global handlers before test callbacks.GlobalHandlers = nil // Create test handlers handler1 := NewHandlerBuilder(). OnStartFn(func(ctx context.Context, info *RunInfo, input CallbackInput) context.Context { return ctx }).Build() handler2 := NewHandlerBuilder(). OnEndFn(func(ctx context.Context, info *RunInfo, output CallbackOutput) context.Context { return ctx }).Build() // Test appending first handler AppendGlobalHandlers(handler1) assert.Equal(t, 1, len(callbacks.GlobalHandlers)) assert.Contains(t, callbacks.GlobalHandlers, handler1) // Test appending second handler AppendGlobalHandlers(handler2) assert.Equal(t, 2, len(callbacks.GlobalHandlers)) assert.Contains(t, callbacks.GlobalHandlers, handler1) assert.Contains(t, callbacks.GlobalHandlers, handler2) // Test appending nil handler AppendGlobalHandlers([]Handler{}...) assert.Equal(t, 2, len(callbacks.GlobalHandlers)) } ``` ## /components/document/callback_extra_loader.go ```go path="/components/document/callback_extra_loader.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package document import ( "github.com/cloudwego/eino/callbacks" "github.com/cloudwego/eino/schema" ) // LoaderCallbackInput is the input for the loader callback. type LoaderCallbackInput struct { // Source is the source of the documents. Source Source // Extra is the extra information for the callback. Extra map[string]any } // LoaderCallbackOutput is the output for the loader callback. type LoaderCallbackOutput struct { // Source is the source of the documents. Source Source // Docs is the documents to be loaded. Docs []*schema.Document // Extra is the extra information for the callback. Extra map[string]any } // ConvLoaderCallbackInput converts the callback input to the loader callback input. func ConvLoaderCallbackInput(src callbacks.CallbackInput) *LoaderCallbackInput { switch t := src.(type) { case *LoaderCallbackInput: return t case Source: return &LoaderCallbackInput{ Source: t, } default: return nil } } // ConvLoaderCallbackOutput converts the callback output to the loader callback output. func ConvLoaderCallbackOutput(src callbacks.CallbackOutput) *LoaderCallbackOutput { switch t := src.(type) { case *LoaderCallbackOutput: return t case []*schema.Document: return &LoaderCallbackOutput{ Docs: t, } default: return nil } } ``` ## /components/document/callback_extra_transformer.go ```go path="/components/document/callback_extra_transformer.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package document import ( "github.com/cloudwego/eino/callbacks" "github.com/cloudwego/eino/schema" ) // TransformerCallbackInput is the input for the transformer callback. type TransformerCallbackInput struct { // Input is the input documents. Input []*schema.Document // Extra is the extra information for the callback. Extra map[string]any } // TransformerCallbackOutput is the output for the transformer callback. type TransformerCallbackOutput struct { // Output is the output documents. Output []*schema.Document // Extra is the extra information for the callback. Extra map[string]any } // ConvTransformerCallbackInput converts the callback input to the transformer callback input. func ConvTransformerCallbackInput(src callbacks.CallbackInput) *TransformerCallbackInput { switch t := src.(type) { case *TransformerCallbackInput: return t case []*schema.Document: return &TransformerCallbackInput{ Input: t, } default: return nil } } // ConvTransformerCallbackOutput converts the callback output to the transformer callback output. func ConvTransformerCallbackOutput(src callbacks.CallbackOutput) *TransformerCallbackOutput { switch t := src.(type) { case *TransformerCallbackOutput: return t case []*schema.Document: return &TransformerCallbackOutput{ Output: t, } default: return nil } } ``` ## /components/document/doc.go ```go path="/components/document/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package document ``` ## /components/document/interface.go ```go path="/components/document/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package document import ( "context" "github.com/cloudwego/eino/schema" ) // Source is a document source. // e.g. https://www.bytedance.com/docx/xxxx, https://xxx.xxx.xxx/xx.pdf. // make sure the URI can be reached by service. type Source struct { URI string } //go:generate mockgen -destination ../../internal/mock/components/document/document_mock.go --package document -source interface.go // Loader is a document loader. type Loader interface { Load(ctx context.Context, src Source, opts ...LoaderOption) ([]*schema.Document, error) } // Transformer is to convert documents, such as split or filter. type Transformer interface { Transform(ctx context.Context, src []*schema.Document, opts ...TransformerOption) ([]*schema.Document, error) } ``` ## /components/document/option.go ```go path="/components/document/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package document // LoaderOption defines call option for Loader component, which is part of the component interface signature. // Each Loader implementation could define its own options struct and option funcs within its own package, // then wrap the impl specific option funcs into this type, before passing to Load. type LoaderOption struct { implSpecificOptFn any } // WrapLoaderImplSpecificOptFn wraps the impl specific option functions into LoaderOption type. // T: the type of the impl specific options struct. // Loader implementations are required to use this function to convert its own option functions into the unified LoaderOption type. // For example, if the Loader impl defines its own options struct: // // type customOptions struct { // conf string // } // // Then the impl needs to provide an option function as such: // // func WithConf(conf string) Option { // return WrapLoaderImplSpecificOptFn(func(o *customOptions) { // o.conf = conf // } // } func WrapLoaderImplSpecificOptFn[T any](optFn func(*T)) LoaderOption { return LoaderOption{ implSpecificOptFn: optFn, } } // GetLoaderImplSpecificOptions provides Loader author the ability to extract their own custom options from the unified LoaderOption type. // T: the type of the impl specific options struct. // This function should be used within the Loader implementation's Load function. // It is recommended to provide a base T as the first argument, within which the Loader author can provide default values for the impl specific options. // eg. // // myOption := &MyOption{ // Field1: "default_value", // } // myOption := loader.GetLoaderImplSpecificOptions(myOption, opts...) func GetLoaderImplSpecificOptions[T any](base *T, opts ...LoaderOption) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { s, ok := opt.implSpecificOptFn.(func(*T)) if ok { s(base) } } } return base } // TransformerOption defines call option for Transformer component, which is part of the component interface signature. // Each Transformer implementation could define its own options struct and option funcs within its own package, // then wrap the impl specific option funcs into this type, before passing to Transform. type TransformerOption struct { implSpecificOptFn any } // WrapTransformerImplSpecificOptFn wraps the impl specific option functions into TransformerOption type. // T: the type of the impl specific options struct. // Transformer implementations are required to use this function to convert its own option functions into the unified TransformerOption type. // For example, if the Transformer impl defines its own options struct: // // type customOptions struct { // conf string // } // // Then the impl needs to provide an option function as such: // // func WithConf(conf string) TransformerOption { // return WrapTransformerImplSpecificOptFn(func(o *customOptions) { // o.conf = conf // } // } // // . func WrapTransformerImplSpecificOptFn[T any](optFn func(*T)) TransformerOption { return TransformerOption{ implSpecificOptFn: optFn, } } // GetTransformerImplSpecificOptions provides Transformer author the ability to extract their own custom options from the unified TransformerOption type. // T: the type of the impl specific options struct. // This function should be used within the Transformer implementation's Transform function. // It is recommended to provide a base T as the first argument, within which the Transformer author can provide default values for the impl specific options. // eg. // // myOption := &MyOption{ // Field1: "default_value", // } // myOption := transformer.GetTransformerImplSpecificOptions(myOption, opts...) func GetTransformerImplSpecificOptions[T any](base *T, opts ...TransformerOption) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { s, ok := opt.implSpecificOptFn.(func(*T)) if ok { s(base) } } } return base } ``` ## /components/document/option_test.go ```go path="/components/document/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package document import ( "testing" "github.com/smartystreets/goconvey/convey" ) func TestImplSpecificOpts(t *testing.T) { type implSpecificOptions struct { conf string index int } withConf := func(conf string) func(o *implSpecificOptions) { return func(o *implSpecificOptions) { o.conf = conf } } withIndex := func(index int) func(o *implSpecificOptions) { return func(o *implSpecificOptions) { o.index = index } } convey.Convey("TestLoaderImplSpecificOpts", t, func() { documentOption1 := WrapLoaderImplSpecificOptFn(withConf("test_conf")) documentOption2 := WrapLoaderImplSpecificOptFn(withIndex(1)) implSpecificOpts := GetLoaderImplSpecificOptions(&implSpecificOptions{}, documentOption1, documentOption2) convey.So(implSpecificOpts, convey.ShouldResemble, &implSpecificOptions{ conf: "test_conf", index: 1, }) }) convey.Convey("TestTransformerImplSpecificOpts", t, func() { documentOption1 := WrapTransformerImplSpecificOptFn(withConf("test_conf")) documentOption2 := WrapTransformerImplSpecificOptFn(withIndex(1)) implSpecificOpts := GetTransformerImplSpecificOptions(&implSpecificOptions{}, documentOption1, documentOption2) convey.So(implSpecificOpts, convey.ShouldResemble, &implSpecificOptions{ conf: "test_conf", index: 1, }) }) } ``` ## /components/document/parser/ext_parser.go ```go path="/components/document/parser/ext_parser.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package parser import ( "context" "errors" "io" "path/filepath" "github.com/cloudwego/eino/schema" ) // ExtParserConfig defines the configuration for the ExtParser. type ExtParserConfig struct { // ext -> parser. // eg: map[string]Parser{ // ".pdf": &PDFParser{}, // ".md": &MarkdownParser{}, // } Parsers map[string]Parser // Fallback parser to use when no other parser is found. // Default is TextParser if not set. FallbackParser Parser } // ExtParser is a parser that uses the file extension to determine which parser to use. // You can register your own parsers by calling RegisterParser. // Default parser is TextParser. // Note: // // parse 时，是通过 filepath.Ext(uri) 的方式找到对应的 parser，因此使用时需要： // ① 必须使用 parser.WithURI 在请求时传入 URI // ② URI 必须能通过 filepath.Ext 来解析出符合预期的 ext // // eg: // // pdf, _ := os.Open("./testdata/test.pdf") // docs, err := ExtParser.Parse(ctx, pdf, parser.WithURI("./testdata/test.pdf")) type ExtParser struct { parsers map[string]Parser fallbackParser Parser } // NewExtParser creates a new ExtParser. func NewExtParser(ctx context.Context, conf *ExtParserConfig) (*ExtParser, error) { if conf == nil { conf = &ExtParserConfig{} } p := &ExtParser{ parsers: conf.Parsers, fallbackParser: conf.FallbackParser, } if p.fallbackParser == nil { p.fallbackParser = TextParser{} } if p.parsers == nil { p.parsers = make(map[string]Parser) } return p, nil } // GetParsers returns a copy of the registered parsers. // It is safe to modify the returned parsers. func (p *ExtParser) GetParsers() map[string]Parser { res := make(map[string]Parser, len(p.parsers)) for k, v := range p.parsers { res[k] = v } return res } // Parse parses the given reader and returns a list of documents. func (p *ExtParser) Parse(ctx context.Context, reader io.Reader, opts ...Option) ([]*schema.Document, error) { opt := GetCommonOptions(&Options{}, opts...) ext := filepath.Ext(opt.URI) parser, ok := p.parsers[ext] if !ok { parser = p.fallbackParser } if parser == nil { return nil, errors.New("no parser found for extension " + ext) } docs, err := parser.Parse(ctx, reader, opts...) if err != nil { return nil, err } for _, doc := range docs { if doc == nil { continue } if doc.MetaData == nil { doc.MetaData = make(map[string]any) } for k, v := range opt.ExtraMeta { doc.MetaData[k] = v } } return docs, nil } ``` ## /components/document/parser/interface.go ```go path="/components/document/parser/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package parser import ( "context" "io" "github.com/cloudwego/eino/schema" ) // Parser is a document parser, can be used to parse a document from a reader. type Parser interface { Parse(ctx context.Context, reader io.Reader, opts ...Option) ([]*schema.Document, error) } ``` ## /components/document/parser/option.go ```go path="/components/document/parser/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package parser type Options struct { // uri of source. URI string // extra metadata will merge to each document. ExtraMeta map[string]any } // Option defines call option for Parser component, which is part of the component interface signature. // Each Parser implementation could define its own options struct and option funcs within its own package, // then wrap the impl specific option funcs into this type, before passing to Transform. type Option struct { apply func(opts *Options) implSpecificOptFn any } // WithURI specifies the URI of the document. // It will be used as to select parser in ExtParser. func WithURI(uri string) Option { return Option{ apply: func(opts *Options) { opts.URI = uri }, } } // WithExtraMeta specifies the extra meta data of the document. func WithExtraMeta(meta map[string]any) Option { return Option{ apply: func(opts *Options) { opts.ExtraMeta = meta }, } } // GetCommonOptions extract parser Options from Option list, optionally providing a base Options with default values. func GetCommonOptions(base *Options, opts ...Option) *Options { if base == nil { base = &Options{} } for i := range opts { opt := opts[i] if opt.apply != nil { opt.apply(base) } } return base } // WrapImplSpecificOptFn wraps the impl specific option functions into Option type. // T: the type of the impl specific options struct. // Parser implementations are required to use this function to convert its own option functions into the unified Option type. // For example, if the Parser impl defines its own options struct: // // type customOptions struct { // conf string // } // // Then the impl needs to provide an option function as such: // // func WithConf(conf string) Option { // return WrapImplSpecificOptFn(func(o *customOptions) { // o.conf = conf // } // } // // . func WrapImplSpecificOptFn[T any](optFn func(*T)) Option { return Option{ implSpecificOptFn: optFn, } } // GetImplSpecificOptions provides Parser author the ability to extract their own custom options from the unified Option type. // T: the type of the impl specific options struct. // This function should be used within the Parser implementation's Transform function. // It is recommended to provide a base T as the first argument, within which the Parser author can provide default values for the impl specific options. func GetImplSpecificOptions[T any](base *T, opts ...Option) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { s, ok := opt.implSpecificOptFn.(func(*T)) if ok { s(base) } } } return base } ``` ## /components/document/parser/option_test.go ```go path="/components/document/parser/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package parser import ( "testing" "github.com/smartystreets/goconvey/convey" ) func TestImplSpecificOpts(t *testing.T) { type implSpecificOptions struct { conf string index int } withConf := func(conf string) func(o *implSpecificOptions) { return func(o *implSpecificOptions) { o.conf = conf } } withIndex := func(index int) func(o *implSpecificOptions) { return func(o *implSpecificOptions) { o.index = index } } convey.Convey("TestImplSpecificOpts", t, func() { parserOption1 := WrapImplSpecificOptFn(withConf("test_conf")) parserOption2 := WrapImplSpecificOptFn(withIndex(1)) implSpecificOpts := GetImplSpecificOptions(&implSpecificOptions{}, parserOption1, parserOption2) convey.So(implSpecificOpts, convey.ShouldResemble, &implSpecificOptions{ conf: "test_conf", index: 1, }) }) } ``` ## /components/document/parser/parser_test.go ```go path="/components/document/parser/parser_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package parser import ( "context" "io" "os" "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/schema" ) type ParserForTest struct { mock func() ([]*schema.Document, error) } func (p *ParserForTest) Parse(ctx context.Context, reader io.Reader, opts ...Option) ([]*schema.Document, error) { return p.mock() } func TestParser(t *testing.T) { ctx := context.Background() t.Run("Test default parser", func(t *testing.T) { conf := &ExtParserConfig{} p, err := NewExtParser(ctx, conf) if err != nil { t.Fatal(err) } f, err := os.Open("testdata/test.md") if err != nil { t.Fatal(err) } defer f.Close() docs, err := p.Parse(ctx, f, WithURI("testdata/test.md")) if err != nil { t.Fatal(err) } assert.Equal(t, 1, len(docs)) assert.Equal(t, "# Title\nhello world", docs[0].Content) }) t.Run("test types", func(t *testing.T) { mockParser := &ParserForTest{ mock: func() ([]*schema.Document, error) { return []*schema.Document{ { Content: "hello world", MetaData: map[string]any{ "type": "text", }, }, }, nil }, } conf := &ExtParserConfig{ Parsers: map[string]Parser{ ".md": mockParser, }, } p, err := NewExtParser(ctx, conf) if err != nil { t.Fatal(err) } f, err := os.Open("testdata/test.md") if err != nil { t.Fatal(err) } defer f.Close() docs, err := p.Parse(ctx, f, WithURI("x/test.md")) if err != nil { t.Fatal(err) } assert.Equal(t, 1, len(docs)) assert.Equal(t, "hello world", docs[0].Content) assert.Equal(t, "text", docs[0].MetaData["type"]) }) t.Run("test get parsers", func(t *testing.T) { p, err := NewExtParser(ctx, &ExtParserConfig{ Parsers: map[string]Parser{ ".md": &TextParser{}, }, }) if err != nil { t.Fatal(err) } ps := p.GetParsers() assert.Equal(t, 1, len(ps)) }) } ``` ## /components/document/parser/testdata/test.md # Title hello world ## /components/document/parser/text_parser.go ```go path="/components/document/parser/text_parser.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package parser import ( "context" "io" "github.com/cloudwego/eino/schema" ) const ( MetaKeySource = "_source" ) // TextParser is a simple parser that reads the text from a reader and returns a single document. // eg: // // docs, err := TextParser.Parse(ctx, strings.NewReader("hello world")) // fmt.Println(docs[0].Content) // "hello world" type TextParser struct{} // Parse reads the text from a reader and returns a single document. func (dp TextParser) Parse(ctx context.Context, reader io.Reader, opts ...Option) ([]*schema.Document, error) { data, err := io.ReadAll(reader) if err != nil { return nil, err } opt := GetCommonOptions(&Options{}, opts...) meta := make(map[string]any) meta[MetaKeySource] = opt.URI for k, v := range opt.ExtraMeta { meta[k] = v } doc := &schema.Document{ Content: string(data), MetaData: meta, } return []*schema.Document{doc}, nil } ``` ## /components/embedding/callback_extra.go ```go path="/components/embedding/callback_extra.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package embedding import ( "github.com/cloudwego/eino/callbacks" ) // TokenUsage is the token usage for the embedding. type TokenUsage struct { // PromptTokens is the number of prompt tokens. PromptTokens int // CompletionTokens is the number of completion tokens. CompletionTokens int // TotalTokens is the total number of tokens. TotalTokens int } // Config is the config for the embedding. type Config struct { // Model is the model name. Model string // EncodingFormat is the encoding format. EncodingFormat string } // ComponentExtra is the extra information for the embedding. type ComponentExtra struct { // Config is the config for the embedding. Config *Config // TokenUsage is the token usage for the embedding. TokenUsage *TokenUsage } // CallbackInput is the input for the embedding callback. type CallbackInput struct { // Texts is the texts to be embedded. Texts []string // Config is the config for the embedding. Config *Config // Extra is the extra information for the callback. Extra map[string]any } // CallbackOutput is the output for the embedding callback. type CallbackOutput struct { // Embeddings is the embeddings. Embeddings [][]float64 // Config is the config for creating the embedding. Config *Config // TokenUsage is the token usage for the embedding. TokenUsage *TokenUsage // Extra is the extra information for the callback. Extra map[string]any } // ConvCallbackInput converts the callback input to the embedding callback input. func ConvCallbackInput(src callbacks.CallbackInput) *CallbackInput { switch t := src.(type) { case *CallbackInput: return t case []string: return &CallbackInput{ Texts: t, } default: return nil } } // ConvCallbackOutput converts the callback output to the embedding callback output. func ConvCallbackOutput(src callbacks.CallbackOutput) *CallbackOutput { switch t := src.(type) { case *CallbackOutput: return t case [][]float64: return &CallbackOutput{ Embeddings: t, } default: return nil } } ``` ## /components/embedding/callback_extra_test.go ```go path="/components/embedding/callback_extra_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package embedding import ( "testing" "github.com/stretchr/testify/assert" ) func TestConvEmbedding(t *testing.T) { assert.NotNil(t, ConvCallbackInput(&CallbackInput{})) assert.NotNil(t, ConvCallbackInput([]string{})) assert.Nil(t, ConvCallbackInput("asd")) assert.NotNil(t, ConvCallbackOutput(&CallbackOutput{})) assert.NotNil(t, ConvCallbackOutput([][]float64{})) assert.Nil(t, ConvCallbackOutput("asd")) } ``` ## /components/embedding/doc.go ```go path="/components/embedding/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package embedding ``` ## /components/embedding/interface.go ```go path="/components/embedding/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package embedding import "context" //go:generate mockgen -destination ../../internal/mock/components/embedding/Embedding_mock.go --package embedding -source interface.go type Embedder interface { EmbedStrings(ctx context.Context, texts []string, opts ...Option) ([][]float64, error) // invoke } ``` ## /components/embedding/option.go ```go path="/components/embedding/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package embedding // Options is the options for the embedding. type Options struct { // Model is the model name for the embedding. Model *string } // Option is the call option for Embedder component. type Option struct { apply func(opts *Options) implSpecificOptFn any } // WithModel is the option to set the model for the embedding. func WithModel(model string) Option { return Option{ apply: func(opts *Options) { opts.Model = &model }, } } // GetCommonOptions extract embedding Options from Option list, optionally providing a base Options with default values. // eg. // // defaultModelName := "default_model" // embeddingOption := &embedding.Options{ // Model: &defaultModelName, // } // embeddingOption := embedding.GetCommonOptions(embeddingOption, opts...) func GetCommonOptions(base *Options, opts ...Option) *Options { if base == nil { base = &Options{} } for i := range opts { opt := opts[i] if opt.apply != nil { opt.apply(base) } } return base } // WrapImplSpecificOptFn is the option to wrap the implementation specific option function. func WrapImplSpecificOptFn[T any](optFn func(*T)) Option { return Option{ implSpecificOptFn: optFn, } } // GetImplSpecificOptions extract the implementation specific options from Option list, optionally providing a base options with default values. // e.g. // // myOption := &MyOption{ // Field1: "default_value", // } // // myOption := model.GetImplSpecificOptions(myOption, opts...) func GetImplSpecificOptions[T any](base *T, opts ...Option) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { optFn, ok := opt.implSpecificOptFn.(func(*T)) if ok { optFn(base) } } } return base } ``` ## /components/embedding/option_test.go ```go path="/components/embedding/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package embedding import ( "testing" "github.com/stretchr/testify/assert" ) func TestOptions(t *testing.T) { defaultModel := "default_model" opts := GetCommonOptions(&Options{Model: &defaultModel}, WithModel("test_model")) assert.NotNil(t, opts.Model) assert.Equal(t, *opts.Model, "test_model") } ``` ## /components/indexer/callback_extra.go ```go path="/components/indexer/callback_extra.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package indexer import ( "github.com/cloudwego/eino/callbacks" "github.com/cloudwego/eino/schema" ) // CallbackInput is the input for the indexer callback. type CallbackInput struct { // Docs is the documents to be indexed. Docs []*schema.Document // Extra is the extra information for the callback. Extra map[string]any } // CallbackOutput is the output for the indexer callback. type CallbackOutput struct { // IDs is the ids of the indexed documents returned by the indexer. IDs []string // Extra is the extra information for the callback. Extra map[string]any } // ConvCallbackInput converts the callback input to the indexer callback input. func ConvCallbackInput(src callbacks.CallbackInput) *CallbackInput { switch t := src.(type) { case *CallbackInput: return t case []*schema.Document: return &CallbackInput{ Docs: t, } default: return nil } } // ConvCallbackOutput converts the callback output to the indexer callback output. func ConvCallbackOutput(src callbacks.CallbackOutput) *CallbackOutput { switch t := src.(type) { case *CallbackOutput: return t case []string: return &CallbackOutput{ IDs: t, } default: return nil } } ``` ## /components/indexer/callback_extra_test.go ```go path="/components/indexer/callback_extra_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package indexer import ( "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/schema" ) func TestConvIndexer(t *testing.T) { assert.NotNil(t, ConvCallbackInput(&CallbackInput{})) assert.NotNil(t, ConvCallbackInput([]*schema.Document{})) assert.Nil(t, ConvCallbackInput("asd")) assert.NotNil(t, ConvCallbackOutput(&CallbackOutput{})) assert.NotNil(t, ConvCallbackOutput([]string{})) assert.Nil(t, ConvCallbackOutput("asd")) } ``` ## /components/indexer/doc.go ```go path="/components/indexer/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package indexer ``` ## /components/indexer/interface.go ```go path="/components/indexer/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package indexer import ( "context" "github.com/cloudwego/eino/schema" ) // Indexer is the interface for the indexer. // Indexer is used to store the documents. // //go:generate mockgen -destination ../../internal/mock/components/indexer/indexer_mock.go --package indexer -source interface.go type Indexer interface { // Store stores the documents. Store(ctx context.Context, docs []*schema.Document, opts ...Option) (ids []string, err error) // invoke } ``` ## /components/indexer/option.go ```go path="/components/indexer/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package indexer import "github.com/cloudwego/eino/components/embedding" // Options is the options for the indexer. type Options struct { // SubIndexes is the sub indexes to be indexed. SubIndexes []string // Embedding is the embedding component. Embedding embedding.Embedder } // WithSubIndexes is the option to set the sub indexes for the indexer. func WithSubIndexes(subIndexes []string) Option { return Option{ apply: func(opts *Options) { opts.SubIndexes = subIndexes }, } } // WithEmbedding is the option to set the embedder for the indexer, which convert document to embeddings. func WithEmbedding(emb embedding.Embedder) Option { return Option{ apply: func(opts *Options) { opts.Embedding = emb }, } } // Option is the call option for Indexer component. type Option struct { apply func(opts *Options) implSpecificOptFn any } // GetCommonOptions extract indexer Options from Option list, optionally providing a base Options with default values. // e.g. // // indexerOption := &IndexerOption{ // SubIndexes: []string{"default_sub_index"}, // default value // } // // indexerOption := indexer.GetCommonOptions(indexerOption, opts...) func GetCommonOptions(base *Options, opts ...Option) *Options { if base == nil { base = &Options{} } for i := range opts { opt := opts[i] if opt.apply != nil { opt.apply(base) } } return base } // WrapImplSpecificOptFn is the option to wrap the implementation specific option function. func WrapImplSpecificOptFn[T any](optFn func(*T)) Option { return Option{ implSpecificOptFn: optFn, } } // GetImplSpecificOptions extract the implementation specific options from Option list, optionally providing a base options with default values. // e.g. // // myOption := &MyOption{ // Field1: "default_value", // } // // myOption := model.GetImplSpecificOptions(myOption, opts...) func GetImplSpecificOptions[T any](base *T, opts ...Option) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { optFn, ok := opt.implSpecificOptFn.(func(*T)) if ok { optFn(base) } } } return base } ``` ## /components/indexer/option_test.go ```go path="/components/indexer/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package indexer import ( "testing" "github.com/smartystreets/goconvey/convey" "github.com/cloudwego/eino/internal/mock/components/embedding" ) func TestOptions(t *testing.T) { convey.Convey("test options", t, func() { var ( subIndexes = []string{"index_1", "index_2"} e = &embedding.MockEmbedder{} ) opts := GetCommonOptions( &Options{}, WithSubIndexes(subIndexes), WithEmbedding(e), ) convey.So(opts, convey.ShouldResemble, &Options{ SubIndexes: subIndexes, Embedding: e, }) }) } ``` ## /components/model/callback_extra.go ```go path="/components/model/callback_extra.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package model import ( "github.com/cloudwego/eino/callbacks" "github.com/cloudwego/eino/schema" ) // TokenUsage is the token usage for the model. type TokenUsage struct { // PromptTokens is the number of prompt tokens. PromptTokens int // CompletionTokens is the number of completion tokens. CompletionTokens int // TotalTokens is the total number of tokens. TotalTokens int } // Config is the config for the model. type Config struct { // Model is the model name. Model string // MaxTokens is the max number of tokens, if reached the max tokens, the model will stop generating, and mostly return an finish reason of "length". MaxTokens int // Temperature is the temperature, which controls the randomness of the model. Temperature float32 // TopP is the top p, which controls the diversity of the model. TopP float32 // Stop is the stop words, which controls the stopping condition of the model. Stop []string } // CallbackInput is the input for the model callback. type CallbackInput struct { // Messages is the messages to be sent to the model. Messages []*schema.Message // Tools is the tools to be used in the model. Tools []*schema.ToolInfo // ToolChoice is the tool choice, which controls the tool to be used in the model. // Deprecated: ToolChoice is no longer supported and should not be set. ToolChoice any // string / *schema.ToolInfo // Config is the config for the model. Config *Config // Extra is the extra information for the callback. Extra map[string]any } // CallbackOutput is the output for the model callback. type CallbackOutput struct { // Message is the message generated by the model. Message *schema.Message // Config is the config for the model. Config *Config // TokenUsage is the token usage of this request. TokenUsage *TokenUsage // Extra is the extra information for the callback. Extra map[string]any } // ConvCallbackInput converts the callback input to the model callback input. func ConvCallbackInput(src callbacks.CallbackInput) *CallbackInput { switch t := src.(type) { case *CallbackInput: // when callback is triggered within component implementation, the input is usually already a typed *model.CallbackInput return t case []*schema.Message: // when callback is injected by graph node, not the component implementation itself, the input is the input of Chat Model interface, which is []*schema.Message return &CallbackInput{ Messages: t, } default: return nil } } // ConvCallbackOutput converts the callback output to the model callback output. func ConvCallbackOutput(src callbacks.CallbackOutput) *CallbackOutput { switch t := src.(type) { case *CallbackOutput: // when callback is triggered within component implementation, the output is usually already a typed *model.CallbackOutput return t case *schema.Message: // when callback is injected by graph node, not the component implementation itself, the output is the output of Chat Model interface, which is *schema.Message return &CallbackOutput{ Message: t, } default: return nil } } ``` ## /components/model/callback_extra_test.go ```go path="/components/model/callback_extra_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package model import ( "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/schema" ) func TestConvModel(t *testing.T) { assert.NotNil(t, ConvCallbackInput(&CallbackInput{})) assert.NotNil(t, ConvCallbackInput([]*schema.Message{})) assert.Nil(t, ConvCallbackInput("asd")) assert.NotNil(t, ConvCallbackOutput(&CallbackOutput{})) assert.NotNil(t, ConvCallbackOutput(&schema.Message{})) assert.Nil(t, ConvCallbackOutput("asd")) } ``` ## /components/model/doc.go ```go path="/components/model/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package model ``` ## /components/model/interface.go ```go path="/components/model/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package model import ( "context" "github.com/cloudwego/eino/schema" ) // BaseChatModel defines the basic interface for chat models. // It provides methods for generating complete outputs and streaming outputs. // This interface serves as the foundation for all chat model implementations. // //go:generate mockgen -destination ../../internal/mock/components/model/ChatModel_mock.go --package model -source interface.go type BaseChatModel interface { Generate(ctx context.Context, input []*schema.Message, opts ...Option) (*schema.Message, error) Stream(ctx context.Context, input []*schema.Message, opts ...Option) ( *schema.StreamReader[*schema.Message], error) } // Deprecated: Please use ToolCallingChatModel interface instead, which provides a safer way to bind tools // without the concurrency issues and tool overwriting problems that may arise from the BindTools method. type ChatModel interface { BaseChatModel // BindTools bind tools to the model. // BindTools before requesting ChatModel generally. // notice the non-atomic problem of BindTools and Generate. BindTools(tools []*schema.ToolInfo) error } // ToolCallingChatModel extends BaseChatModel with tool calling capabilities. // It provides a WithTools method that returns a new instance with // the specified tools bound, avoiding state mutation and concurrency issues. type ToolCallingChatModel interface { BaseChatModel // WithTools returns a new ToolCallingChatModel instance with the specified tools bound. // This method does not modify the current instance, making it safer for concurrent use. WithTools(tools []*schema.ToolInfo) (ToolCallingChatModel, error) } ``` ## /components/model/option.go ```go path="/components/model/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package model import "github.com/cloudwego/eino/schema" // Options is the common options for the model. type Options struct { // Temperature is the temperature for the model, which controls the randomness of the model. Temperature *float32 // MaxTokens is the max number of tokens, if reached the max tokens, the model will stop generating, and mostly return an finish reason of "length". MaxTokens *int // Model is the model name. Model *string // TopP is the top p for the model, which controls the diversity of the model. TopP *float32 // Stop is the stop words for the model, which controls the stopping condition of the model. Stop []string // Tools is a list of tools the model may call. Tools []*schema.ToolInfo // ToolChoice controls which tool is called by the model. ToolChoice *schema.ToolChoice } // Option is the call option for ChatModel component. type Option struct { apply func(opts *Options) implSpecificOptFn any } // WithTemperature is the option to set the temperature for the model. func WithTemperature(temperature float32) Option { return Option{ apply: func(opts *Options) { opts.Temperature = &temperature }, } } // WithMaxTokens is the option to set the max tokens for the model. func WithMaxTokens(maxTokens int) Option { return Option{ apply: func(opts *Options) { opts.MaxTokens = &maxTokens }, } } // WithModel is the option to set the model name. func WithModel(name string) Option { return Option{ apply: func(opts *Options) { opts.Model = &name }, } } // WithTopP is the option to set the top p for the model. func WithTopP(topP float32) Option { return Option{ apply: func(opts *Options) { opts.TopP = &topP }, } } // WithStop is the option to set the stop words for the model. func WithStop(stop []string) Option { return Option{ apply: func(opts *Options) { opts.Stop = stop }, } } // WithTools is the option to set tools for the model. func WithTools(tools []*schema.ToolInfo) Option { if tools == nil { tools = []*schema.ToolInfo{} } return Option{ apply: func(opts *Options) { opts.Tools = tools }, } } // WithToolChoice is the option to set tool choice for the model. func WithToolChoice(toolChoice schema.ToolChoice) Option { return Option{ apply: func(opts *Options) { opts.ToolChoice = &toolChoice }, } } // WrapImplSpecificOptFn is the option to wrap the implementation specific option function. func WrapImplSpecificOptFn[T any](optFn func(*T)) Option { return Option{ implSpecificOptFn: optFn, } } // GetCommonOptions extract model Options from Option list, optionally providing a base Options with default values. func GetCommonOptions(base *Options, opts ...Option) *Options { if base == nil { base = &Options{} } for i := range opts { opt := opts[i] if opt.apply != nil { opt.apply(base) } } return base } // GetImplSpecificOptions extract the implementation specific options from Option list, optionally providing a base options with default values. // e.g. // // myOption := &MyOption{ // Field1: "default_value", // } // // myOption := model.GetImplSpecificOptions(myOption, opts...) func GetImplSpecificOptions[T any](base *T, opts ...Option) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { optFn, ok := opt.implSpecificOptFn.(func(*T)) if ok { optFn(base) } } } return base } ``` ## /components/model/option_test.go ```go path="/components/model/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package model import ( "testing" "github.com/cloudwego/eino/schema" "github.com/smartystreets/goconvey/convey" ) func TestOptions(t *testing.T) { convey.Convey("test options", t, func() { var ( modelName = "model" temperature float32 = 0.9 maxToken = 5000 topP float32 = 0.8 defaultModel = "default_model" defaultTemperature float32 = 1.0 defaultMaxTokens = 1000 defaultTopP float32 = 0.5 tools = []*schema.ToolInfo{{Name: "asd"}, {Name: "qwe"}} toolChoice = schema.ToolChoiceForced ) opts := GetCommonOptions( &Options{ Model: &defaultModel, Temperature: &defaultTemperature, MaxTokens: &defaultMaxTokens, TopP: &defaultTopP, }, WithModel(modelName), WithTemperature(temperature), WithMaxTokens(maxToken), WithTopP(topP), WithStop([]string{"hello", "bye"}), WithTools(tools), WithToolChoice(toolChoice), ) convey.So(opts, convey.ShouldResemble, &Options{ Model: &modelName, Temperature: &temperature, MaxTokens: &maxToken, TopP: &topP, Stop: []string{"hello", "bye"}, Tools: tools, ToolChoice: &toolChoice, }) }) convey.Convey("test nil tools option", t, func() { opts := GetCommonOptions( &Options{ Tools: []*schema.ToolInfo{ {Name: "asd"}, {Name: "qwe"}, }, }, WithTools(nil), ) convey.So(opts.Tools, convey.ShouldNotBeNil) convey.So(len(opts.Tools), convey.ShouldEqual, 0) }) } type implOption struct { userID int64 name string } func WithUserID(uid int64) Option { return WrapImplSpecificOptFn[implOption](func(i *implOption) { i.userID = uid }) } func WithName(n string) Option { return WrapImplSpecificOptFn[implOption](func(i *implOption) { i.name = n }) } func TestImplSpecificOption(t *testing.T) { convey.Convey("impl_specific_option", t, func() { opt := GetImplSpecificOptions(&implOption{}, WithUserID(101), WithName("Wang")) convey.So(opt, convey.ShouldEqual, &implOption{ userID: 101, name: "Wang", }) }) } ``` ## /components/prompt/callback_extra.go ```go path="/components/prompt/callback_extra.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt import ( "github.com/cloudwego/eino/callbacks" "github.com/cloudwego/eino/schema" ) // CallbackInput is the input for the callback. type CallbackInput struct { // Variables is the variables for the callback. Variables map[string]any // Templates is the templates for the callback. Templates []schema.MessagesTemplate // Extra is the extra information for the callback. Extra map[string]any } // CallbackOutput is the output for the callback. type CallbackOutput struct { // Result is the result for the callback. Result []*schema.Message // Templates is the templates for the callback. Templates []schema.MessagesTemplate // Extra is the extra information for the callback. Extra map[string]any } // ConvCallbackInput converts the callback input to the prompt callback input. func ConvCallbackInput(src callbacks.CallbackInput) *CallbackInput { switch t := src.(type) { case *CallbackInput: return t case map[string]any: return &CallbackInput{ Variables: t, } default: return nil } } // ConvCallbackOutput converts the callback output to the prompt callback output. func ConvCallbackOutput(src callbacks.CallbackOutput) *CallbackOutput { switch t := src.(type) { case *CallbackOutput: return t case []*schema.Message: return &CallbackOutput{ Result: t, } default: return nil } } ``` ## /components/prompt/callback_extra_test.go ```go path="/components/prompt/callback_extra_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt import ( "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/schema" ) func TestConvPrompt(t *testing.T) { assert.NotNil(t, ConvCallbackInput(&CallbackInput{})) assert.NotNil(t, ConvCallbackInput(map[string]any{})) assert.Nil(t, ConvCallbackInput("asd")) assert.NotNil(t, ConvCallbackOutput(&CallbackOutput{})) assert.NotNil(t, ConvCallbackOutput([]*schema.Message{})) assert.Nil(t, ConvCallbackOutput("asd")) } ``` ## /components/prompt/chat_template.go ```go path="/components/prompt/chat_template.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt import ( "context" "github.com/cloudwego/eino/callbacks" "github.com/cloudwego/eino/schema" ) // DefaultChatTemplate is the default chat template implementation. type DefaultChatTemplate struct { // templates is the templates for the chat template. templates []schema.MessagesTemplate // formatType is the format type for the chat template. formatType schema.FormatType } // FromMessages creates a new DefaultChatTemplate from the given templates and format type. // eg. // // template := prompt.FromMessages(schema.FString, &schema.Message{Content: "Hello, {name}!"}, &schema.Message{Content: "how are you?"}) // // in chain, or graph // chain := compose.NewChain[map[string]any, []*schema.Message]() // chain.AppendChatTemplate(template) func FromMessages(formatType schema.FormatType, templates ...schema.MessagesTemplate) *DefaultChatTemplate { return &DefaultChatTemplate{ templates: templates, formatType: formatType, } } // Format formats the chat template with the given context and variables. func (t *DefaultChatTemplate) Format(ctx context.Context, vs map[string]any, _ ...Option) (result []*schema.Message, err error) { defer func() { if err != nil { _ = callbacks.OnError(ctx, err) } }() ctx = callbacks.OnStart(ctx, &CallbackInput{ Variables: vs, Templates: t.templates, }) result = make([]*schema.Message, 0, len(t.templates)) for _, template := range t.templates { msgs, err := template.Format(ctx, vs, t.formatType) if err != nil { return nil, err } result = append(result, msgs...) } _ = callbacks.OnEnd(ctx, &CallbackOutput{ Result: result, Templates: t.templates, }) return result, nil } // GetType returns the type of the chat template (Default). func (t *DefaultChatTemplate) GetType() string { return "Default" } // IsCallbacksEnabled checks if the callbacks are enabled for the chat template. func (t *DefaultChatTemplate) IsCallbacksEnabled() bool { return true } ``` ## /components/prompt/chat_template_test.go ```go path="/components/prompt/chat_template_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt import ( "context" "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/schema" ) func TestFormat(t *testing.T) { pyFmtTestTemplate := []schema.MessagesTemplate{ schema.SystemMessage( "you are a helpful assistant.\n" + "here is the context: {context}"), schema.MessagesPlaceholder("chat_history", true), schema.UserMessage("question: {question}"), } jinja2TestTemplate := []schema.MessagesTemplate{ schema.SystemMessage( "you are a helpful assistant.\n" + "here is the context: {{context}}"), schema.MessagesPlaceholder("chat_history", true), schema.UserMessage("question: {{question}}"), } goFmtTestTemplate := []schema.MessagesTemplate{ schema.SystemMessage( "you are a helpful assistant.\n" + "here is the context: {{.context}}"), schema.MessagesPlaceholder("chat_history", true), schema.UserMessage("question: {{.question}}"), } testValues := map[string]any{ "context": "it's beautiful day", "question": "how is the day today", "chat_history": []*schema.Message{ schema.UserMessage("who are you"), schema.AssistantMessage("I'm a helpful assistant", nil), }, } expected := []*schema.Message{ schema.SystemMessage( "you are a helpful assistant.\n" + "here is the context: it's beautiful day"), schema.UserMessage("who are you"), schema.AssistantMessage("I'm a helpful assistant", nil), schema.UserMessage("question: how is the day today"), } // FString chatTemplate := FromMessages(schema.FString, pyFmtTestTemplate...) msgs, err := chatTemplate.Format(context.Background(), testValues) assert.Nil(t, err) assert.Equal(t, expected, msgs) // Jinja2 chatTemplate = FromMessages(schema.Jinja2, jinja2TestTemplate...) msgs, err = chatTemplate.Format(context.Background(), testValues) assert.Nil(t, err) assert.Equal(t, expected, msgs) // GoTemplate chatTemplate = FromMessages(schema.GoTemplate, goFmtTestTemplate...) msgs, err = chatTemplate.Format(context.Background(), testValues) assert.Nil(t, err) assert.Equal(t, expected, msgs) } func TestDocumentFormat(t *testing.T) { docs := []*schema.Document{ { ID: "1", Content: "qwe", MetaData: map[string]any{ "hello": 888, }, }, { ID: "2", Content: "asd", MetaData: map[string]any{ "bye": 111, }, }, } template := FromMessages(schema.FString, schema.SystemMessage("all:{all_docs}\nsingle:{single_doc}"), ) msgs, err := template.Format(context.Background(), map[string]any{ "all_docs": docs, "single_doc": docs[0], }) assert.Nil(t, err) t.Log(msgs) } ``` ## /components/prompt/doc.go ```go path="/components/prompt/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt ``` ## /components/prompt/interface.go ```go path="/components/prompt/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt import ( "context" "github.com/cloudwego/eino/schema" ) var _ ChatTemplate = &DefaultChatTemplate{} type ChatTemplate interface { Format(ctx context.Context, vs map[string]any, opts ...Option) ([]*schema.Message, error) } ``` ## /components/prompt/option.go ```go path="/components/prompt/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt // Option is the call option for ChatTemplate component. type Option struct { implSpecificOptFn any } // WrapImplSpecificOptFn wraps the implementation specific option function. func WrapImplSpecificOptFn[T any](optFn func(*T)) Option { return Option{ implSpecificOptFn: optFn, } } // GetImplSpecificOptions extracts the implementation specific options from Option list, optionally providing a base options with default values. func GetImplSpecificOptions[T any](base *T, opts ...Option) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { s, ok := opt.implSpecificOptFn.(func(*T)) if ok { s(base) } } } return base } ``` ## /components/prompt/option_test.go ```go path="/components/prompt/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package prompt import ( "testing" "github.com/smartystreets/goconvey/convey" ) type implOption struct { userID int64 name string } func WithUserID(uid int64) Option { return WrapImplSpecificOptFn[implOption](func(i *implOption) { i.userID = uid }) } func WithName(n string) Option { return WrapImplSpecificOptFn[implOption](func(i *implOption) { i.name = n }) } func TestImplSpecificOption(t *testing.T) { convey.Convey("impl_specific_option", t, func() { opt := GetImplSpecificOptions(&implOption{}, WithUserID(101), WithName("Wang")) convey.So(opt, convey.ShouldEqual, &implOption{ userID: 101, name: "Wang", }) }) } ``` ## /components/retriever/callback_extra.go ```go path="/components/retriever/callback_extra.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package retriever import ( "github.com/cloudwego/eino/callbacks" "github.com/cloudwego/eino/schema" ) // CallbackInput is the input for the retriever callback. type CallbackInput struct { // Query is the query for the retriever. Query string // TopK is the top k for the retriever, which means the top number of documents to retrieve. TopK int // Filter is the filter for the retriever. Filter string // ScoreThreshold is the score threshold for the retriever, eg 0.5 means the score of the document must be greater than 0.5. ScoreThreshold *float64 // Extra is the extra information for the retriever. Extra map[string]any } // CallbackOutput is the output for the retriever callback. type CallbackOutput struct { // Docs is the documents for the retriever. Docs []*schema.Document // Extra is the extra information for the retriever. Extra map[string]any } // ConvCallbackInput converts the callback input to the retriever callback input. func ConvCallbackInput(src callbacks.CallbackInput) *CallbackInput { switch t := src.(type) { case *CallbackInput: return t case string: return &CallbackInput{ Query: t, } default: return nil } } // ConvCallbackOutput converts the callback output to the retriever callback output. func ConvCallbackOutput(src callbacks.CallbackOutput) *CallbackOutput { switch t := src.(type) { case *CallbackOutput: return t case []*schema.Document: return &CallbackOutput{ Docs: t, } default: return nil } } ``` ## /components/retriever/callback_extra_test.go ```go path="/components/retriever/callback_extra_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package retriever import ( "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/schema" ) func TestConvRetriever(t *testing.T) { assert.NotNil(t, ConvCallbackInput(&CallbackInput{})) assert.NotNil(t, ConvCallbackInput("asd")) assert.Nil(t, ConvCallbackInput([]string{})) assert.NotNil(t, ConvCallbackOutput(&CallbackOutput{})) assert.NotNil(t, ConvCallbackOutput([]*schema.Document{})) assert.Nil(t, ConvCallbackOutput("asd")) } ``` ## /components/retriever/doc.go ```go path="/components/retriever/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package retriever ``` ## /components/retriever/interface.go ```go path="/components/retriever/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package retriever import ( "context" "github.com/cloudwego/eino/schema" ) //go:generate mockgen -destination ../../internal/mock/components/retriever/retriever_mock.go --package retriever -source interface.go // Retriever is the interface for retriever. // It is used to retrieve documents from a source. // // e.g. // // retriever, err := redis.NewRetriever(ctx, &redis.RetrieverConfig{}) // if err != nil {...} // docs, err := retriever.Retrieve(ctx, "query") // <= using directly // docs, err := retriever.Retrieve(ctx, "query", retriever.WithTopK(3)) // <= using options // // graph := compose.NewGraph[inputType, outputType](compose.RunTypeDAG) // graph.AddRetrieverNode("retriever_node_key", retriever) // <= using in graph type Retriever interface { Retrieve(ctx context.Context, query string, opts ...Option) ([]*schema.Document, error) } ``` ## /components/retriever/option.go ```go path="/components/retriever/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package retriever import "github.com/cloudwego/eino/components/embedding" // Options is the options for the retriever. type Options struct { // Index is the index for the retriever, index in different retriever may be different. Index *string // SubIndex is the sub index for the retriever, sub index in different retriever may be different. SubIndex *string // TopK is the top k for the retriever, which means the top number of documents to retrieve. TopK *int // ScoreThreshold is the score threshold for the retriever, eg 0.5 means the score of the document must be greater than 0.5. ScoreThreshold *float64 // Embedding is the embedder for the retriever, which is used to embed the query for retrieval . Embedding embedding.Embedder // DSLInfo is the dsl info for the retriever, which is used to retrieve the documents from the retriever. // viking only DSLInfo map[string]interface{} } // WithIndex wraps the index option. func WithIndex(index string) Option { return Option{ apply: func(opts *Options) { opts.Index = &index }, } } // WithSubIndex wraps the sub index option. func WithSubIndex(subIndex string) Option { return Option{ apply: func(opts *Options) { opts.SubIndex = &subIndex }, } } // WithTopK wraps the top k option. func WithTopK(topK int) Option { return Option{ apply: func(opts *Options) { opts.TopK = &topK }, } } // WithScoreThreshold wraps the score threshold option. func WithScoreThreshold(threshold float64) Option { return Option{ apply: func(opts *Options) { opts.ScoreThreshold = &threshold }, } } // WithEmbedding wraps the embedder option. func WithEmbedding(emb embedding.Embedder) Option { return Option{ apply: func(opts *Options) { opts.Embedding = emb }, } } // WithDSLInfo wraps the dsl info option. func WithDSLInfo(dsl map[string]any) Option { return Option{ apply: func(opts *Options) { opts.DSLInfo = dsl }, } } // Option is the call option for Retriever component. type Option struct { apply func(opts *Options) implSpecificOptFn any } // GetCommonOptions extract retriever Options from Option list, optionally providing a base Options with default values. func GetCommonOptions(base *Options, opts ...Option) *Options { if base == nil { base = &Options{} } for i := range opts { if opts[i].apply != nil { opts[i].apply(base) } } return base } // WrapImplSpecificOptFn is the option to wrap the implementation specific option function. func WrapImplSpecificOptFn[T any](optFn func(*T)) Option { return Option{ implSpecificOptFn: optFn, } } // GetImplSpecificOptions extract the implementation specific options from Option list, optionally providing a base options with default values. // e.g. // // myOption := &MyOption{ // Field1: "default_value", // } // // myOption := model.GetImplSpecificOptions(myOption, opts...) func GetImplSpecificOptions[T any](base *T, opts ...Option) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { optFn, ok := opt.implSpecificOptFn.(func(*T)) if ok { optFn(base) } } } return base } ``` ## /components/retriever/option_test.go ```go path="/components/retriever/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package retriever import ( "testing" "github.com/smartystreets/goconvey/convey" "github.com/cloudwego/eino/internal/mock/components/embedding" ) func TestOptions(t *testing.T) { convey.Convey("test options", t, func() { var ( index = "index" topK = 2 scoreThreshold = 4.0 subIndex = "sub_index" dslInfo = map[string]any{"dsl": "dsl"} e = &embedding.MockEmbedder{} defaultTopK = 1 ) opts := GetCommonOptions( &Options{ TopK: &defaultTopK, }, WithIndex(index), WithTopK(topK), WithScoreThreshold(scoreThreshold), WithSubIndex(subIndex), WithDSLInfo(dslInfo), WithEmbedding(e), ) convey.So(opts, convey.ShouldResemble, &Options{ Index: &index, TopK: &topK, ScoreThreshold: &scoreThreshold, SubIndex: &subIndex, DSLInfo: dslInfo, Embedding: e, }) }) } ``` ## /components/tool/callback_extra.go ```go path="/components/tool/callback_extra.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package tool import ( "github.com/cloudwego/eino/callbacks" ) // CallbackInput is the input for the tool callback. type CallbackInput struct { // ArgumentsInJSON is the arguments in json format for the tool. ArgumentsInJSON string // Extra is the extra information for the tool. Extra map[string]any } // CallbackOutput is the output for the tool callback. type CallbackOutput struct { // Response is the response for the tool. Response string // Extra is the extra information for the tool. Extra map[string]any } // ConvCallbackInput converts the callback input to the tool callback input. func ConvCallbackInput(src callbacks.CallbackInput) *CallbackInput { switch t := src.(type) { case *CallbackInput: return t case string: return &CallbackInput{ArgumentsInJSON: t} default: return nil } } // ConvCallbackOutput converts the callback output to the tool callback output. func ConvCallbackOutput(src callbacks.CallbackOutput) *CallbackOutput { switch t := src.(type) { case *CallbackOutput: return t case string: return &CallbackOutput{Response: t} default: return nil } } ``` ## /components/tool/callback_extra_test.go ```go path="/components/tool/callback_extra_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package tool import ( "testing" "github.com/stretchr/testify/assert" ) func TestConvCallbackInput(t *testing.T) { assert.NotNil(t, ConvCallbackInput(&CallbackInput{})) assert.NotNil(t, ConvCallbackInput("asd")) assert.Nil(t, ConvCallbackInput(123)) assert.Nil(t, ConvCallbackInput(nil)) } func TestConvCallbackOutput(t *testing.T) { assert.NotNil(t, ConvCallbackOutput(&CallbackOutput{})) assert.NotNil(t, ConvCallbackOutput("asd")) assert.Nil(t, ConvCallbackOutput(123)) assert.Nil(t, ConvCallbackOutput(nil)) } ``` ## /components/tool/doc.go ```go path="/components/tool/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package tool ``` ## /components/tool/interface.go ```go path="/components/tool/interface.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package tool import ( "context" "github.com/cloudwego/eino/schema" ) // BaseTool get tool info for ChatModel intent recognition. type BaseTool interface { Info(ctx context.Context) (*schema.ToolInfo, error) } // InvokableTool the tool for ChatModel intent recognition and ToolsNode execution. // nolint: byted_s_interface_name type InvokableTool interface { BaseTool // InvokableRun call function with arguments in JSON format InvokableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (string, error) } // StreamableTool the stream tool for ChatModel intent recognition and ToolsNode execution. // nolint: byted_s_interface_name type StreamableTool interface { BaseTool StreamableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (*schema.StreamReader[string], error) } ``` ## /components/tool/option.go ```go path="/components/tool/option.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package tool // Option defines call option for InvokableTool or StreamableTool component, which is part of component interface signature. // Each tool implementation could define its own options struct and option funcs within its own package, // then wrap the impl specific option funcs into this type, before passing to InvokableRun or StreamableRun. type Option struct { implSpecificOptFn any } // WrapImplSpecificOptFn wraps the impl specific option functions into Option type. // T: the type of the impl specific options struct. // Tool implementations are required to use this function to convert its own option functions into the unified Option type. // For example, if the tool defines its own options struct: // // type customOptions struct { // conf string // } // // Then the tool needs to provide an option function as such: // // func WithConf(conf string) Option { // return WrapImplSpecificOptFn(func(o *customOptions) { // o.conf = conf // } // } // // . func WrapImplSpecificOptFn[T any](optFn func(*T)) Option { return Option{ implSpecificOptFn: optFn, } } // GetImplSpecificOptions provides tool author the ability to extract their own custom options from the unified Option type. // T: the type of the impl specific options struct. // This function should be used within the tool implementation's InvokableRun or StreamableRun functions. // It is recommended to provide a base T as the first argument, within which the tool author can provide default values for the impl specific options. // eg. // // type customOptions struct { // conf string // } // defaultOptions := &customOptions{} // // customOptions := tool.GetImplSpecificOptions(defaultOptions, opts...) func GetImplSpecificOptions[T any](base *T, opts ...Option) *T { if base == nil { base = new(T) } for i := range opts { opt := opts[i] if opt.implSpecificOptFn != nil { optFn, ok := opt.implSpecificOptFn.(func(*T)) if ok { optFn(base) } } } return base } ``` ## /components/tool/option_test.go ```go path="/components/tool/option_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package tool import ( "testing" "github.com/smartystreets/goconvey/convey" ) func TestImplSpecificOpts(t *testing.T) { convey.Convey("TestImplSpecificOpts", t, func() { type implSpecificOptions struct { conf string index int } withConf := func(conf string) func(o *implSpecificOptions) { return func(o *implSpecificOptions) { o.conf = conf } } withIndex := func(index int) func(o *implSpecificOptions) { return func(o *implSpecificOptions) { o.index = index } } toolOption1 := WrapImplSpecificOptFn(withConf("test_conf")) toolOption2 := WrapImplSpecificOptFn(withIndex(1)) implSpecificOpts := GetImplSpecificOptions(&implSpecificOptions{}, toolOption1, toolOption2) convey.So(implSpecificOpts, convey.ShouldResemble, &implSpecificOptions{ conf: "test_conf", index: 1, }) }) } ``` ## /components/tool/utils/create_options.go ```go path="/components/tool/utils/create_options.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils import ( "context" "fmt" "reflect" "sort" "strings" "github.com/getkin/kin-openapi/openapi3" ) // UnmarshalArguments is the function type for unmarshalling the arguments. type UnmarshalArguments func(ctx context.Context, arguments string) (interface{}, error) // MarshalOutput is the function type for marshalling the output. type MarshalOutput func(ctx context.Context, output interface{}) (string, error) type toolOptions struct { um UnmarshalArguments m MarshalOutput sc SchemaCustomizerFn } // Option is the option func for the tool. type Option func(o *toolOptions) // WithUnmarshalArguments wraps the unmarshal arguments option. // when you want to unmarshal the arguments by yourself, you can use this option. func WithUnmarshalArguments(um UnmarshalArguments) Option { return func(o *toolOptions) { o.um = um } } // WithMarshalOutput wraps the marshal output option. // when you want to marshal the output by yourself, you can use this option. func WithMarshalOutput(m MarshalOutput) Option { return func(o *toolOptions) { o.m = m } } // SchemaCustomizerFn is the schema customizer function for inferring tool parameter from tagged go struct. // Within this function, end-user can parse custom go struct tags into corresponding openapi schema field. // Parameters: // 1. name: the name of current schema, usually the field name of the go struct. Specifically, the last 'name' visited is fixed to be '_root', which represents the entire go struct. Also, for array field, both the field itself and the element within the array will trigger this function. // 2. t: the type of current schema, usually the field type of the go struct. // 3. tag: the struct tag of current schema, usually the field tag of the go struct. Note that the element within an array field will use the same go struct tag as the array field itself. // 4. schema: the current openapi schema object to be customized. type SchemaCustomizerFn func(name string, t reflect.Type, tag reflect.StructTag, schema *openapi3.Schema) error // WithSchemaCustomizer sets a user-defined schema customizer for inferring tool parameter from tagged go struct. // If this option is not set, the defaultSchemaCustomizer will be used. func WithSchemaCustomizer(sc SchemaCustomizerFn) Option { return func(o *toolOptions) { o.sc = sc } } func getToolOptions(opt ...Option) *toolOptions { opts := &toolOptions{ um: nil, m: nil, } for _, o := range opt { o(opts) } return opts } // defaultSchemaCustomizer is the default schema customizer when using reflect to infer tool parameter from tagged go struct. // Supported struct tags: // 1. jsonschema: "description=xxx" // 2. jsonschema: "enum=xxx,enum=yyy,enum=zzz" // 3. jsonschema: "required" // 4. can also use json: "xxx,omitempty" to mark the field as not required, which means an absence of 'omitempty' in json tag means the field is required. // If this defaultSchemaCustomizer is not sufficient or suitable to your specific need, define your own SchemaCustomizerFn and pass it to WithSchemaCustomizer during InferTool or InferStreamTool. func defaultSchemaCustomizer(name string, t reflect.Type, tag reflect.StructTag, schema *openapi3.Schema) error { jsonS := tag.Get("jsonschema") if len(jsonS) > 0 { tags := strings.Split(jsonS, ",") for _, t := range tags { kv := strings.Split(t, "=") if len(kv) == 2 { if kv[0] == "description" { schema.Description = kv[1] } if kv[0] == "enum" { schema.Enum = append(schema.Enum, kv[1]) } } else if len(kv) == 1 { if kv[0] == "required" { if schema.Extensions == nil { schema.Extensions = make(map[string]any, 1) } schema.Extensions["x_required"] = true } } } } json := tag.Get("json") if len(json) > 0 && !strings.Contains(json, "omitempty") { if schema.Extensions == nil { schema.Extensions = make(map[string]any, 1) } schema.Extensions["x_required"] = true } if name == "_root" { if err := setRequired(schema); err != nil { return err } } return nil } func setRequired(sc *openapi3.Schema) error { // check if properties are marked as required, set schema required to true accordingly if sc.Type != openapi3.TypeObject && sc.Type != openapi3.TypeArray { return nil } if sc.Type == openapi3.TypeArray { if sc.Items.Value.Extensions != nil { if _, ok := sc.Items.Value.Extensions["x_required"]; ok { delete(sc.Items.Value.Extensions, "x_required") if len(sc.Items.Value.Extensions) == 0 { sc.Items.Value.Extensions = nil } } } if err := setRequired(sc.Items.Value); err != nil { return fmt.Errorf("setRequired for array failed: %w", err) } } for k, p := range sc.Properties { if p.Value.Extensions != nil { if _, ok := p.Value.Extensions["x_required"]; ok { sc.Required = append(sc.Required, k) delete(p.Value.Extensions, "x_required") if len(p.Value.Extensions) == 0 { p.Value.Extensions = nil } } } err := setRequired(p.Value) if err != nil { return fmt.Errorf("setRequired for nested property %s failed: %w", k, err) } } sort.Strings(sc.Required) return nil } ``` ## /components/tool/utils/doc.go ```go path="/components/tool/utils/doc.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils ``` ## /components/tool/utils/error_handler.go ```go path="/components/tool/utils/error_handler.go" /* * Copyright 2025 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils import ( "context" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" ) type ErrorHandler func(context.Context, error) string // WrapToolWithErrorHandler wraps any BaseTool with custom error handling. // This function detects the tool type (InvokableTool, StreamableTool, or both) // and applies the appropriate error handling wrapper. // When the wrapped tool returns an error, the error handler function 'h' will be called // to convert the error into a string result, and no error will be returned from the wrapper. // // Parameters: // - t: The original BaseTool to be wrapped // - h: A function that converts an error to a string // // Returns: // - A wrapped BaseTool that handles errors internally based on its capabilities func WrapToolWithErrorHandler(t tool.BaseTool, h ErrorHandler) tool.BaseTool { ih := &infoHelper{info: t.Info} var s tool.StreamableTool if st, ok := t.(tool.StreamableTool); ok { s = st } if it, ok := t.(tool.InvokableTool); ok { if s == nil { return WrapInvokableToolWithErrorHandler(it, h) } else { return &combinedErrorWrapper{ infoHelper: ih, errorHelper: &errorHelper{ i: it.InvokableRun, h: h, }, streamErrorHelper: &streamErrorHelper{ s: s.StreamableRun, h: h, }, } } } if s != nil { return WrapStreamableToolWithErrorHandler(s, h) } return t } // WrapInvokableToolWithErrorHandler wraps an InvokableTool with custom error handling. // When the wrapped tool returns an error, the error handler function 'h' will be called // to convert the error into a string result, and no error will be returned from the wrapper. // // Parameters: // - tool: The original InvokableTool to be wrapped // - h: A function that converts an error to a string // // Returns: // - A wrapped InvokableTool that handles errors internally func WrapInvokableToolWithErrorHandler(t tool.InvokableTool, h ErrorHandler) tool.InvokableTool { return &errorWrapper{ infoHelper: &infoHelper{info: t.Info}, errorHelper: &errorHelper{ i: t.InvokableRun, h: h, }, } } // WrapStreamableToolWithErrorHandler wraps a StreamableTool with custom error handling. // When the wrapped tool returns an error, the error handler function 'h' will be called // to convert the error into a string result, which will be returned as a single-item stream, // and no error will be returned from the wrapper. // // Parameters: // - tool: The original StreamableTool to be wrapped // - h: A function that converts an error to a string // // Returns: // - A wrapped StreamableTool that handles errors internally func WrapStreamableToolWithErrorHandler(t tool.StreamableTool, h ErrorHandler) tool.StreamableTool { return &streamErrorWrapper{ infoHelper: &infoHelper{info: t.Info}, streamErrorHelper: &streamErrorHelper{ s: t.StreamableRun, h: h, }, } } type errorWrapper struct { *infoHelper *errorHelper } type streamErrorWrapper struct { *infoHelper *streamErrorHelper } type combinedErrorWrapper struct { *infoHelper *errorHelper *streamErrorHelper } type infoHelper struct { info func(ctx context.Context) (*schema.ToolInfo, error) } func (i *infoHelper) Info(ctx context.Context) (*schema.ToolInfo, error) { return i.info(ctx) } type errorHelper struct { i func(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (string, error) h ErrorHandler } func (s *errorHelper) InvokableRun(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (string, error) { result, err := s.i(ctx, argumentsInJSON, opts...) if err != nil { return s.h(ctx, err), nil } return result, nil } type streamErrorHelper struct { s func(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (*schema.StreamReader[string], error) h ErrorHandler } func (s *streamErrorHelper) StreamableRun(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (*schema.StreamReader[string], error) { result, err := s.s(ctx, argumentsInJSON, opts...) if err != nil { return schema.StreamReaderFromArray([]string{s.h(ctx, err)}), nil } return result, nil } ``` ## /components/tool/utils/error_handler_test.go ```go path="/components/tool/utils/error_handler_test.go" /* * Copyright 2025 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils import ( "context" "errors" "io" "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" ) type testErrorTool struct{} func (t *testErrorTool) Info(ctx context.Context) (*schema.ToolInfo, error) { return nil, nil } func (t *testErrorTool) InvokableRun(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (string, error) { return "", errors.New("test error") } func (t *testErrorTool) StreamableRun(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (*schema.StreamReader[string], error) { return nil, errors.New("test stream error") } func TestErrorWrapper(t *testing.T) { ctx := context.Background() nt := WrapToolWithErrorHandler(&testErrorTool{}, func(_ context.Context, err error) string { return err.Error() }) result, err := nt.(tool.InvokableTool).InvokableRun(ctx, "") assert.NoError(t, err) assert.Equal(t, "test error", result) streamResult, err := nt.(tool.StreamableTool).StreamableRun(ctx, "") assert.NoError(t, err) chunk, err := streamResult.Recv() assert.NoError(t, err) assert.Equal(t, "test stream error", chunk) _, err = streamResult.Recv() assert.True(t, errors.Is(err, io.EOF)) wrappedTool := WrapInvokableToolWithErrorHandler(&testErrorTool{}, func(_ context.Context, err error) string { return err.Error() }) result, err = wrappedTool.InvokableRun(ctx, "") assert.NoError(t, err) assert.Equal(t, "test error", result) wrappedStreamTool := WrapStreamableToolWithErrorHandler(&testErrorTool{}, func(_ context.Context, err error) string { return err.Error() }) streamResult, err = wrappedStreamTool.StreamableRun(ctx, "") assert.NoError(t, err) chunk, err = streamResult.Recv() assert.NoError(t, err) assert.Equal(t, "test stream error", chunk) _, err = streamResult.Recv() assert.True(t, errors.Is(err, io.EOF)) } ``` ## /components/tool/utils/invokable_func.go ```go path="/components/tool/utils/invokable_func.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils import ( "context" "fmt" "strings" "github.com/bytedance/sonic" "github.com/getkin/kin-openapi/openapi3gen" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/internal/generic" "github.com/cloudwego/eino/schema" ) // InvokeFunc is the function type for the tool. type InvokeFunc[T, D any] func(ctx context.Context, input T) (output D, err error) // OptionableInvokeFunc is the function type for the tool with tool option. type OptionableInvokeFunc[T, D any] func(ctx context.Context, input T, opts ...tool.Option) (output D, err error) // InferTool creates an InvokableTool from a given function by inferring the ToolInfo from the function's request parameters. // End-user can pass a SchemaCustomizerFn in opts to customize the go struct tag parsing process, overriding default behavior. func InferTool[T, D any](toolName, toolDesc string, i InvokeFunc[T, D], opts ...Option) (tool.InvokableTool, error) { ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...) if err != nil { return nil, err } return NewTool(ti, i, opts...), nil } // InferOptionableTool creates an InvokableTool from a given function by inferring the ToolInfo from the function's request parameters, with tool option. func InferOptionableTool[T, D any](toolName, toolDesc string, i OptionableInvokeFunc[T, D], opts ...Option) (tool.InvokableTool, error) { ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...) if err != nil { return nil, err } return newOptionableTool(ti, i, opts...), nil } // GoStruct2ParamsOneOf converts a go struct to a ParamsOneOf. // if you attempt to use ResponseFormat of some ChatModel to get StructuredOutput, you can infer the JSONSchema from the go struct. func GoStruct2ParamsOneOf[T any](opts ...Option) (*schema.ParamsOneOf, error) { return goStruct2ParamsOneOf[T](opts...) } // GoStruct2ToolInfo converts a go struct to a ToolInfo. // if you attempt to use BindTool to make ChatModel respond StructuredOutput, you can infer the ToolInfo from the go struct. func GoStruct2ToolInfo[T any](toolName, toolDesc string, opts ...Option) (*schema.ToolInfo, error) { return goStruct2ToolInfo[T](toolName, toolDesc, opts...) } func goStruct2ToolInfo[T any](toolName, toolDesc string, opts ...Option) (*schema.ToolInfo, error) { paramsOneOf, err := goStruct2ParamsOneOf[T](opts...) if err != nil { return nil, err } return &schema.ToolInfo{ Name: toolName, Desc: toolDesc, ParamsOneOf: paramsOneOf, }, nil } func goStruct2ParamsOneOf[T any](opts ...Option) (*schema.ParamsOneOf, error) { options := getToolOptions(opts...) schemaCustomizer := defaultSchemaCustomizer if options.sc != nil { schemaCustomizer = options.sc } sc, err := openapi3gen.NewSchemaRefForValue(generic.NewInstance[T](), nil, openapi3gen.SchemaCustomizer(schemaCustomizer)) if err != nil { return nil, fmt.Errorf("new SchemaRef from T failed: %w", err) } paramsOneOf := schema.NewParamsOneOfByOpenAPIV3(sc.Value) return paramsOneOf, nil } // NewTool Create a tool, where the input and output are both in JSON format. func NewTool[T, D any](desc *schema.ToolInfo, i InvokeFunc[T, D], opts ...Option) tool.InvokableTool { return newOptionableTool(desc, func(ctx context.Context, input T, _ ...tool.Option) (D, error) { return i(ctx, input) }, opts...) } // NewTool Create a tool, where the input and output are both in JSON format. func newOptionableTool[T, D any](desc *schema.ToolInfo, i OptionableInvokeFunc[T, D], opts ...Option) tool.InvokableTool { to := getToolOptions(opts...) return &invokableTool[T, D]{ info: desc, um: to.um, m: to.m, Fn: i, } } type invokableTool[T, D any] struct { info *schema.ToolInfo um UnmarshalArguments m MarshalOutput Fn OptionableInvokeFunc[T, D] } func (i *invokableTool[T, D]) Info(ctx context.Context) (*schema.ToolInfo, error) { return i.info, nil } // InvokableRun invokes the tool with the given arguments. func (i *invokableTool[T, D]) InvokableRun(ctx context.Context, arguments string, opts ...tool.Option) (output string, err error) { var inst T if i.um != nil { var val interface{} val, err = i.um(ctx, arguments) if err != nil { return "", fmt.Errorf("[LocalFunc] failed to unmarshal arguments, toolName=%s, err=%w", i.getToolName(), err) } gt, ok := val.(T) if !ok { return "", fmt.Errorf("[LocalFunc] invalid type, toolName=%s, expected=%T, given=%T", i.getToolName(), inst, val) } inst = gt } else { inst = generic.NewInstance[T]() err = sonic.UnmarshalString(arguments, &inst) if err != nil { return "", fmt.Errorf("[LocalFunc] failed to unmarshal arguments in json, toolName=%s, err=%w", i.getToolName(), err) } } resp, err := i.Fn(ctx, inst, opts...) if err != nil { return "", fmt.Errorf("[LocalFunc] failed to invoke tool, toolName=%s, err=%w", i.getToolName(), err) } if i.m != nil { output, err = i.m(ctx, resp) if err != nil { return "", fmt.Errorf("[LocalFunc] failed to marshal output, toolName=%s, err=%w", i.getToolName(), err) } } else { output, err = sonic.MarshalString(resp) if err != nil { return "", fmt.Errorf("[LocalFunc] failed to marshal output in json, toolName=%s, err=%w", i.getToolName(), err) } } return output, nil } func (i *invokableTool[T, D]) GetType() string { return snakeToCamel(i.getToolName()) } func (i *invokableTool[T, D]) getToolName() string { if i.info == nil { return "" } return i.info.Name } // snakeToCamel converts a snake_case string to CamelCase. func snakeToCamel(s string) string { if s == "" { return "" } parts := strings.Split(s, "_") for i := 0; i < len(parts); i++ { if len(parts[i]) > 0 { parts[i] = strings.ToUpper(string(parts[i][0])) + strings.ToLower(parts[i][1:]) } } return strings.Join(parts, "") } ``` ## /components/tool/utils/invokable_func_test.go ```go path="/components/tool/utils/invokable_func_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils import ( "context" "fmt" "testing" "github.com/getkin/kin-openapi/openapi3" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" ) type Job struct { Company string `json:"company" jsonschema:"description=the company where the user works"` Position string `json:"position,omitempty" jsonschema:"description=the position of the user's job"` ServiceLength float32 `json:"service_length,omitempty" jsonschema:"description=the year of user's service"` // 司龄，年 } type Income struct { Source string `json:"source" jsonschema:"description=the source of income"` Amount int `json:"amount" jsonschema:"description=the amount of income"` HasPayTax bool `json:"has_pay_tax" jsonschema:"description=whether the user has paid tax"` Job *Job `json:"job,omitempty" jsonschema:"description=the job of the user when earning this income"` } type User struct { Name string `json:"name" jsonschema:"required,description=the name of the user"` Age int `json:"age" jsonschema:"required,description=the age of the user"` Job *Job `json:"job,omitempty" jsonschema:"description=the job of the user"` Incomes []*Income `json:"incomes" jsonschema:"description=the incomes of the user"` } type UserResult struct { Code int `json:"code"` Msg string `json:"msg"` } var toolInfo = &schema.ToolInfo{ Name: "update_user_info", Desc: "full update user info", ParamsOneOf: schema.NewParamsOneOfByOpenAPIV3( &openapi3.Schema{ Type: openapi3.TypeObject, Required: []string{"age", "incomes", "name"}, Properties: openapi3.Schemas{ "name": { Value: &openapi3.Schema{ Type: openapi3.TypeString, Description: "the name of the user", }, }, "age": { Value: &openapi3.Schema{ Type: openapi3.TypeInteger, Description: "the age of the user", }, }, "job": { Value: &openapi3.Schema{ Type: openapi3.TypeObject, Description: "the job of the user", Required: []string{"company"}, // Nullable: true, Properties: openapi3.Schemas{ "company": { Value: &openapi3.Schema{ Type: openapi3.TypeString, Description: "the company where the user works", }, }, "service_length": { Value: &openapi3.Schema{ Type: openapi3.TypeNumber, Description: "the year of user's service", Format: "float", }, }, "position": { Value: &openapi3.Schema{ Type: openapi3.TypeString, Description: "the position of the user's job", }, }, }, }, }, "incomes": { Value: &openapi3.Schema{ Type: openapi3.TypeArray, Description: "the incomes of the user", Items: &openapi3.SchemaRef{ Value: &openapi3.Schema{ Type: openapi3.TypeObject, Required: []string{"amount", "has_pay_tax", "source"}, Description: "the incomes of the user", // Nullable: true, Properties: openapi3.Schemas{ "source": { Value: &openapi3.Schema{ Type: openapi3.TypeString, Description: "the source of income", }, }, "amount": { Value: &openapi3.Schema{ Type: openapi3.TypeInteger, Description: "the amount of income", }, }, "has_pay_tax": { Value: &openapi3.Schema{ Type: openapi3.TypeBoolean, Description: "whether the user has paid tax", }, }, "job": { Value: &openapi3.Schema{ Type: openapi3.TypeObject, Description: "the job of the user when earning this income", Required: []string{"company"}, // Nullable: true, Properties: openapi3.Schemas{ "company": { Value: &openapi3.Schema{ Type: openapi3.TypeString, Description: "the company where the user works", }, }, "service_length": { Value: &openapi3.Schema{ Type: openapi3.TypeNumber, Description: "the year of user's service", Format: "float", }, }, "position": { Value: &openapi3.Schema{ Type: openapi3.TypeString, Description: "the position of the user's job", }, }, }, }, }, }, AdditionalProperties: openapi3.AdditionalProperties{}, }, }, }, }, }, AdditionalProperties: openapi3.AdditionalProperties{}, }), } func updateUserInfo(ctx context.Context, input *User) (output *UserResult, err error) { return &UserResult{ Code: 200, Msg: fmt.Sprintf("update %v success", input.Name), }, nil } type UserInfoOption struct { Field1 string } func WithUserInfoOption(s string) tool.Option { return tool.WrapImplSpecificOptFn(func(t *UserInfoOption) { t.Field1 = s }) } func updateUserInfoWithOption(_ context.Context, input *User, opts ...tool.Option) (output *UserResult, err error) { baseOption := &UserInfoOption{ Field1: "test_origin", } option := tool.GetImplSpecificOptions(baseOption, opts...) return &UserResult{ Code: 200, Msg: option.Field1, }, nil } func TestInferTool(t *testing.T) { t.Run("invoke_infer_tool", func(t *testing.T) { ctx := context.Background() tl, err := InferTool("update_user_info", "full update user info", updateUserInfo) assert.NoError(t, err) info, err := tl.Info(context.Background()) assert.NoError(t, err) assert.Equal(t, toolInfo, info) content, err := tl.InvokableRun(ctx, `{"name": "bruce lee"}`) assert.NoError(t, err) assert.JSONEq(t, `{"code":200,"msg":"update bruce lee success"}`, content) }) } func TestInferOptionableTool(t *testing.T) { ctx := context.Background() t.Run("invoke_infer_optionable_tool", func(t *testing.T) { tl, err := InferOptionableTool("invoke_infer_optionable_tool", "full update user info", updateUserInfoWithOption) assert.NoError(t, err) content, err := tl.InvokableRun(ctx, `{"name": "bruce lee"}`, WithUserInfoOption("hello world")) assert.NoError(t, err) assert.JSONEq(t, `{"code":200,"msg":"hello world"}`, content) }) } func TestNewTool(t *testing.T) { ctx := context.Background() type Input struct { Name string `json:"name"` } type Output struct { Name string `json:"name"` } t.Run("struct_input_struct_output", func(t *testing.T) { tl := NewTool[Input, Output](nil, func(ctx context.Context, input Input) (output Output, err error) { return Output{ Name: input.Name, }, nil }) _, err := tl.InvokableRun(ctx, `{"name":"test"}`) assert.Nil(t, err) }) t.Run("pointer_input_pointer_output", func(t *testing.T) { tl := NewTool[*Input, *Output](nil, func(ctx context.Context, input *Input) (output *Output, err error) { return &Output{ Name: input.Name, }, nil }) content, err := tl.InvokableRun(ctx, `{"name":"test"}`) assert.NoError(t, err) assert.Equal(t, `{"name":"test"}`, content) }) t.Run("string_input_int64_output", func(t *testing.T) { tl := NewTool(nil, func(ctx context.Context, input string) (output int64, err error) { return 10, nil }) content, err := tl.InvokableRun(ctx, `100`) // json unmarshal must contains double quote if is not json string. assert.Error(t, err) assert.Equal(t, "", content) }) t.Run("string_pointer_input_int64_pointer_output", func(t *testing.T) { tl := NewTool[*string, *int64](nil, func(ctx context.Context, input *string) (output *int64, err error) { n := int64(10) return &n, nil }) content, err := tl.InvokableRun(ctx, `"100"`) assert.NoError(t, err) assert.Equal(t, `10`, content) }) } func TestSnakeToCamel(t *testing.T) { t.Run("normal_case", func(t *testing.T) { assert.Equal(t, "GoogleSearch3", snakeToCamel("google_search_3")) }) t.Run("empty_case", func(t *testing.T) { assert.Equal(t, "", snakeToCamel("")) }) t.Run("single_word_case", func(t *testing.T) { assert.Equal(t, "Google", snakeToCamel("google")) }) t.Run("upper_case", func(t *testing.T) { assert.Equal(t, "HttpHost", snakeToCamel("_HTTP_HOST_")) }) t.Run("underscore_case", func(t *testing.T) { assert.Equal(t, "", snakeToCamel("_")) }) } type testEnumStruct struct { Field1 string `json:"field1" jsonschema:"enum=a,enum=b"` } func TestEnumTag(t *testing.T) { info, err := goStruct2ParamsOneOf[testEnumStruct]() assert.NoError(t, err) s, err := info.ToOpenAPIV3() assert.NoError(t, err) assert.Equal(t, []interface{}{"a", "b"}, s.Properties["field1"].Value.Enum) } ``` ## /components/tool/utils/streamable_func.go ```go path="/components/tool/utils/streamable_func.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils import ( "context" "fmt" "github.com/bytedance/sonic" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/internal/generic" "github.com/cloudwego/eino/schema" ) // StreamFunc is the function type for the streamable tool. type StreamFunc[T, D any] func(ctx context.Context, input T) (output *schema.StreamReader[D], err error) // OptionableStreamFunc is the function type for the streamable tool with tool option. type OptionableStreamFunc[T, D any] func(ctx context.Context, input T, opts ...tool.Option) (output *schema.StreamReader[D], err error) // InferStreamTool creates an StreamableTool from a given function by inferring the ToolInfo from the function's request parameters // End-user can pass a SchemaCustomizerFn in opts to customize the go struct tag parsing process, overriding default behavior. func InferStreamTool[T, D any](toolName, toolDesc string, s StreamFunc[T, D], opts ...Option) (tool.StreamableTool, error) { ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...) if err != nil { return nil, err } return NewStreamTool(ti, s, opts...), nil } // InferStreamTool creates an StreamableTool from a given function by inferring the ToolInfo from the function's request parameters, with tool option. func InferOptionableStreamTool[T, D any](toolName, toolDesc string, s OptionableStreamFunc[T, D], opts ...Option) (tool.StreamableTool, error) { ti, err := goStruct2ToolInfo[T](toolName, toolDesc, opts...) if err != nil { return nil, err } return newOptionableStreamTool(ti, s, opts...), nil } // NewStreamTool Create a streaming tool, where the input and output are both in JSON format. // convert: convert the stream frame to string that could be concatenated to a string. func NewStreamTool[T, D any](desc *schema.ToolInfo, s StreamFunc[T, D], opts ...Option) tool.StreamableTool { return newOptionableStreamTool(desc, func(ctx context.Context, input T, _ ...tool.Option) (output *schema.StreamReader[D], err error) { return s(ctx, input) }, opts...) } func newOptionableStreamTool[T, D any](desc *schema.ToolInfo, s OptionableStreamFunc[T, D], opts ...Option) tool.StreamableTool { to := getToolOptions(opts...) return &streamableTool[T, D]{ info: desc, um: to.um, m: to.m, Fn: s, } } type streamableTool[T, D any] struct { info *schema.ToolInfo um UnmarshalArguments m MarshalOutput Fn OptionableStreamFunc[T, D] } // Info returns the tool info, implement the BaseTool interface. func (s *streamableTool[T, D]) Info(ctx context.Context) (*schema.ToolInfo, error) { return s.info, nil } // StreamableRun invokes the tool with the given arguments, implement the StreamableTool interface. func (s *streamableTool[T, D]) StreamableRun(ctx context.Context, argumentsInJSON string, opts ...tool.Option) ( outStream *schema.StreamReader[string], err error) { var inst T if s.um != nil { var val interface{} val, err = s.um(ctx, argumentsInJSON) if err != nil { return nil, fmt.Errorf("[LocalStreamFunc] failed to unmarshal arguments, toolName=%s, err=%w", s.getToolName(), err) } gt, ok := val.(T) if !ok { return nil, fmt.Errorf("[LocalStreamFunc] type err, toolName=%s, expected=%T, given=%T", s.getToolName(), inst, val) } inst = gt } else { inst = generic.NewInstance[T]() err = sonic.UnmarshalString(argumentsInJSON, &inst) if err != nil { return nil, fmt.Errorf("[LocalStreamFunc] failed to unmarshal arguments in json, toolName=%s, err=%w", s.getToolName(), err) } } streamD, err := s.Fn(ctx, inst, opts...) if err != nil { return nil, err } outStream = schema.StreamReaderWithConvert(streamD, func(d D) (string, error) { var out string var e error if s.m != nil { out, e = s.m(ctx, d) if e != nil { return "", fmt.Errorf("[LocalStreamFunc] failed to marshal output, toolName=%s, err=%w", s.getToolName(), e) } } else { out, e = sonic.MarshalString(d) if e != nil { return "", fmt.Errorf("[LocalStreamFunc] failed to marshal output in json, toolName=%s, err=%w", s.getToolName(), e) } } return out, nil }) return outStream, nil } func (s *streamableTool[T, D]) GetType() string { return snakeToCamel(s.getToolName()) } func (s *streamableTool[T, D]) getToolName() string { if s.info == nil { return "" } return s.info.Name } ``` ## /components/tool/utils/streamable_func_test.go ```go path="/components/tool/utils/streamable_func_test.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package utils import ( "context" "errors" "io" "testing" "github.com/getkin/kin-openapi/openapi3" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" ) func TestNewStreamableTool(t *testing.T) { ctx := context.Background() type Input struct { Name string `json:"name"` } type Output struct { Name string `json:"name"` } t.Run("simple_case", func(t *testing.T) { tl := NewStreamTool[*Input, *Output]( &schema.ToolInfo{ Name: "search_user", Desc: "search user info", ParamsOneOf: schema.NewParamsOneOfByParams( map[string]*schema.ParameterInfo{ "name": { Type: "string", Desc: "user name", }, }), }, func(ctx context.Context, input *Input) (output *schema.StreamReader[*Output], err error) { sr, sw := schema.Pipe[*Output](2) sw.Send(&Output{ Name: input.Name, }, nil) sw.Send(&Output{ Name: "lee", }, nil) sw.Close() return sr, nil }, ) info, err := tl.Info(ctx) assert.NoError(t, err) assert.Equal(t, "search_user", info.Name) js, err := info.ToOpenAPIV3() assert.NoError(t, err) assert.Equal(t, &openapi3.Schema{ Type: openapi3.TypeObject, Properties: map[string]*openapi3.SchemaRef{ "name": { Value: &openapi3.Schema{ Type: openapi3.TypeString, Description: "user name", }, }, }, Required: make([]string, 0), }, js) sr, err := tl.StreamableRun(ctx, `{"name":"xxx"}`) assert.NoError(t, err) defer sr.Close() idx := 0 for { m, err := sr.Recv() if errors.Is(err, io.EOF) { break } assert.NoError(t, err) if idx == 0 { assert.Equal(t, `{"name":"xxx"}`, m) } else { assert.Equal(t, `{"name":"lee"}`, m) } idx++ } assert.Equal(t, 2, idx) }) } type FakeStreamOption struct { Field string } type FakeStreamInferToolInput struct { Field string `json:"field"` } type FakeStreamInferToolOutput struct { Field string `json:"field"` } func FakeWithToolOption(s string) tool.Option { return tool.WrapImplSpecificOptFn(func(t *FakeStreamOption) { t.Field = s }) } func fakeStreamFunc(ctx context.Context, input FakeStreamInferToolInput, opts ...tool.Option) (output *schema.StreamReader[*FakeStreamInferToolOutput], err error) { baseOpt := &FakeStreamOption{ Field: "default_field_value", } option := tool.GetImplSpecificOptions(baseOpt, opts...) return schema.StreamReaderFromArray([]*FakeStreamInferToolOutput{ { Field: option.Field, }, }), nil } func TestInferStreamTool(t *testing.T) { st, err := InferOptionableStreamTool("infer_optionable_stream_tool", "test infer stream tool with option", fakeStreamFunc) assert.Nil(t, err) sr, err := st.StreamableRun(context.Background(), `{"field": "value"}`, FakeWithToolOption("hello world")) assert.Nil(t, err) defer sr.Close() idx := 0 for { m, err := sr.Recv() if errors.Is(err, io.EOF) { break } assert.NoError(t, err) if idx == 0 { assert.JSONEq(t, `{"field":"hello world"}`, m) } } } ``` ## /components/types.go ```go path="/components/types.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ // components are the basic components supported by eino. package components // Typer get the type name of one component's implementation // if Typer exists, the full name of the component instance will be {Typer}{Component} by default // recommend using Camel Case Naming Style for Typer type Typer interface { GetType() string } func GetType(component any) (string, bool) { if typer, ok := component.(Typer); ok { return typer.GetType(), true } return "", false } // Checker tells callback aspect status of component's implementation // When the Checker interface is implemented and returns true, the framework will not start the default aspect. // Instead, the component will decide the callback execution location and the information to be injected. type Checker interface { IsCallbacksEnabled() bool } func IsCallbacksEnabled(i any) bool { if checker, ok := i.(Checker); ok { return checker.IsCallbacksEnabled() } return false } // Component the name of different kinds of components type Component string const ( ComponentOfPrompt Component = "ChatTemplate" ComponentOfChatModel Component = "ChatModel" ComponentOfEmbedding Component = "Embedding" ComponentOfIndexer Component = "Indexer" ComponentOfRetriever Component = "Retriever" ComponentOfLoader Component = "Loader" ComponentOfTransformer Component = "DocumentTransformer" ComponentOfTool Component = "Tool" ) ``` ## /compose/branch.go ```go path="/compose/branch.go" /* * Copyright 2025 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package compose import ( "context" "fmt" "reflect" "github.com/cloudwego/eino/internal/generic" "github.com/cloudwego/eino/schema" ) // GraphBranchCondition is the condition type for the branch. type GraphBranchCondition[T any] func(ctx context.Context, in T) (endNode string, err error) // StreamGraphBranchCondition is the condition type for the stream branch. type StreamGraphBranchCondition[T any] func(ctx context.Context, in *schema.StreamReader[T]) (endNode string, err error) // GraphMultiBranchCondition is the condition type for the multi choice branch. type GraphMultiBranchCondition[T any] func(ctx context.Context, in T) (endNode map[string]bool, err error) // StreamGraphMultiBranchCondition is the condition type for the stream multi choice branch. type StreamGraphMultiBranchCondition[T any] func(ctx context.Context, in *schema.StreamReader[T]) (endNodes map[string]bool, err error) // GraphBranch is the branch type for the graph. // It is used to determine the next node based on the condition. type GraphBranch struct { invoke func(ctx context.Context, input any) (output []string, err error) collect func(ctx context.Context, input streamReader) (output []string, err error) inputType reflect.Type *genericHelper endNodes map[string]bool idx int // used to distinguish branches in parallel noDataFlow bool } // GetEndNode returns the all end nodes of the branch. func (gb *GraphBranch) GetEndNode() map[string]bool { return gb.endNodes } func newGraphBranch[T any](r *runnablePacker[T, []string, any], endNodes map[string]bool) *GraphBranch { return &GraphBranch{ invoke: func(ctx context.Context, input any) (output []string, err error) { nInput, ok := input.(T) if !ok { panic(newUnexpectedInputTypeErr(generic.TypeOf[T](), reflect.TypeOf(input))) } return r.Invoke(ctx, nInput) }, collect: func(ctx context.Context, input streamReader) (output []string, err error) { in, ok := unpackStreamReader[T](input) if !ok { panic(newUnexpectedInputTypeErr(generic.TypeOf[T](), input.getType())) } return r.Collect(ctx, in) }, inputType: generic.TypeOf[T](), genericHelper: newGenericHelper[T, T](), endNodes: endNodes, } } func NewGraphMultiBranch[T any](condition GraphMultiBranchCondition[T], endNodes map[string]bool) *GraphBranch { condRun := func(ctx context.Context, in T, opts ...any) ([]string, error) { ends, err := condition(ctx, in) if err != nil { return nil, err } ret := make([]string, 0, len(ends)) for end := range ends { if !endNodes[end] { return nil, fmt.Errorf("branch invocation returns unintended end node: %s", end) } ret = append(ret, end) } return ret, nil } return newGraphBranch(newRunnablePacker(condRun, nil, nil, nil, false), endNodes) } func NewStreamGraphMultiBranch[T any](condition StreamGraphMultiBranchCondition[T], endNodes map[string]bool) *GraphBranch { condRun := func(ctx context.Context, in *schema.StreamReader[T], opts ...any) ([]string, error) { ends, err := condition(ctx, in) if err != nil { return nil, err } ret := make([]string, 0, len(ends)) for end := range ends { if !endNodes[end] { return nil, fmt.Errorf("branch invocation returns unintended end node: %s", end) } ret = append(ret, end) } return ret, nil } return newGraphBranch(newRunnablePacker(nil, nil, condRun, nil, false), endNodes) } // NewGraphBranch creates a new graph branch. // It is used to determine the next node based on the condition. // e.g. // // condition := func(ctx context.Context, in string) (string, error) { // // logic to determine the next node // return "next_node_key", nil // } // endNodes := map[string]bool{"path01": true, "path02": true} // branch := compose.NewGraphBranch(condition, endNodes) // // graph.AddBranch("key_of_node_before_branch", branch) func NewGraphBranch[T any](condition GraphBranchCondition[T], endNodes map[string]bool) *GraphBranch { return NewGraphMultiBranch(func(ctx context.Context, in T) (endNode map[string]bool, err error) { ret, err := condition(ctx, in) if err != nil { return nil, err } return map[string]bool{ret: true}, nil }, endNodes) } // NewStreamGraphBranch creates a new stream graph branch. // It is used to determine the next node based on the condition of stream input. // e.g. // // condition := func(ctx context.Context, in *schema.StreamReader[T]) (string, error) { // // logic to determine the next node. // // to use the feature of stream, you can use the first chunk to determine the next node. // return "next_node_key", nil // } // endNodes := map[string]bool{"path01": true, "path02": true} // branch := compose.NewStreamGraphBranch(condition, endNodes) // // graph.AddBranch("key_of_node_before_branch", branch) func NewStreamGraphBranch[T any](condition StreamGraphBranchCondition[T], endNodes map[string]bool) *GraphBranch { return NewStreamGraphMultiBranch(func(ctx context.Context, in *schema.StreamReader[T]) (endNode map[string]bool, err error) { ret, err := condition(ctx, in) if err != nil { return nil, err } return map[string]bool{ret: true}, nil }, endNodes) } ``` ## /compose/branch_test.go ```go path="/compose/branch_test.go" /* * Copyright 2025 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package compose import ( "context" "io" "testing" "github.com/stretchr/testify/assert" "github.com/cloudwego/eino/schema" ) func TestMultiBranch(t *testing.T) { g := NewGraph[string, map[string]any]() emptyLambda := InvokableLambda(func(ctx context.Context, input string) (output string, err error) { return input, nil }) err := g.AddLambdaNode("1", emptyLambda, WithOutputKey("1")) assert.NoError(t, err) err = g.AddLambdaNode("2", emptyLambda, WithOutputKey("2")) assert.NoError(t, err) err = g.AddLambdaNode("3", emptyLambda, WithOutputKey("3")) assert.NoError(t, err) err = g.AddBranch(START, NewGraphMultiBranch(func(ctx context.Context, in string) (endNode map[string]bool, err error) { return map[string]bool{"1": true, "2": true}, nil }, map[string]bool{"1": true, "2": true, "3": true})) assert.NoError(t, err) err = g.AddEdge("1", END) assert.NoError(t, err) err = g.AddEdge("2", END) assert.NoError(t, err) err = g.AddEdge("3", END) assert.NoError(t, err) ctx := context.Background() r, err := g.Compile(ctx) assert.NoError(t, err) result, err := r.Invoke(ctx, "start") assert.NoError(t, err) assert.Equal(t, map[string]any{ "1": "start", "2": "start", }, result) streamResult, err := r.Stream(ctx, "start") assert.NoError(t, err) result = map[string]any{} for { chunk, err := streamResult.Recv() if err == io.EOF { break } assert.NoError(t, err) for k, v := range chunk { result[k] = v } } assert.Equal(t, map[string]any{ "1": "start", "2": "start", }, result) } func TestStreamMultiBranch(t *testing.T) { g := NewGraph[string, map[string]any]() emptyLambda := InvokableLambda(func(ctx context.Context, input string) (output string, err error) { return input, nil }) err := g.AddLambdaNode("1", emptyLambda, WithOutputKey("1")) assert.NoError(t, err) err = g.AddLambdaNode("2", emptyLambda, WithOutputKey("2")) assert.NoError(t, err) err = g.AddLambdaNode("3", emptyLambda, WithOutputKey("3")) assert.NoError(t, err) err = g.AddBranch(START, NewStreamGraphMultiBranch(func(ctx context.Context, in *schema.StreamReader[string]) (endNode map[string]bool, err error) { in.Close() return map[string]bool{"1": true, "2": true}, nil }, map[string]bool{"1": true, "2": true, "3": true})) assert.NoError(t, err) err = g.AddEdge("1", END) assert.NoError(t, err) err = g.AddEdge("2", END) assert.NoError(t, err) err = g.AddEdge("3", END) assert.NoError(t, err) ctx := context.Background() r, err := g.Compile(ctx) assert.NoError(t, err) result, err := r.Invoke(ctx, "start") assert.NoError(t, err) assert.Equal(t, map[string]any{ "1": "start", "2": "start", }, result) streamResult, err := r.Stream(ctx, "start") assert.NoError(t, err) result = map[string]any{} for { chunk, err := streamResult.Recv() if err == io.EOF { break } assert.NoError(t, err) for k, v := range chunk { result[k] = v } } assert.Equal(t, map[string]any{ "1": "start", "2": "start", }, result) } ``` ## /compose/chain.go ```go path="/compose/chain.go" /* * Copyright 2024 CloudWeGo Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package compose import ( "context" "errors" "fmt" "reflect" "github.com/cloudwego/eino/components/document" "github.com/cloudwego/eino/components/embedding" "github.com/cloudwego/eino/components/indexer" "github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/components/prompt" "github.com/cloudwego/eino/components/retriever" "github.com/cloudwego/eino/internal/generic" "github.com/cloudwego/eino/internal/gmap" "github.com/cloudwego/eino/internal/gslice" ) // NewChain create a chain with input/output type. func NewChain[I, O any](opts ...NewGraphOption) *Chain[I, O] { ch := &Chain[I, O]{ gg: NewGraph[I, O](opts...), } ch.gg.cmp = ComponentOfChain return ch } // Chain is a chain of components. // Chain nodes can be parallel / branch / sequence components. // Chain is designed to be used in a builder pattern (should Compile() before use). // And the interface is `Chain style`, you can use it like: `chain.AppendXX(...).AppendXX(...)` // // Normal usage: // 1. create a chain with input/output type: `chain := NewChain[inputType, outputType]()` // 2. add components to chainable list: // 2.1 add components: `chain.AppendChatTemplate(...).AppendChatModel(...).AppendToolsNode(...)` // 2.2 add parallel or branch node if needed: `chain.AppendParallel()`, `chain.AppendBranch()` // 3. compile: `r, err := c.Compile()` // 4. run: // 4.1 `one input & one output` use `r.Invoke(ctx, input)` // 4.2 `one input & multi output chunk` use `r.Stream(ctx, input)` // 4.3 `multi input chunk & one output` use `r.Collect(ctx, inputReader)` // 4.4 `multi input chunk & multi output chunk` use `r.Transform(ctx, inputReader)` // // Using in graph or other chain: // chain1 := NewChain[inputType, outputType]() // graph := NewGraph[](runTypePregel) // graph.AddGraph("key", chain1) // chain is an AnyGraph implementation // // // or in another chain: // chain2 := NewChain[inputType, outputType]() // chain2.AppendGraph(chain1) type Chain[I, O any] struct { err error gg *Graph[I, O] nodeIdx int preNodeKeys []string hasEnd bool } // ErrChainCompiled is returned when attempting to modify a chain after it has been compiled var ErrChainCompiled = errors.New("chain has been compiled, cannot be modified") // implements AnyGraph. func (c *Chain[I, O]) compile(ctx context.Context, option *graphCompileOptions) (*composableRunnable, error) { if err := c.addEndIfNeeded(); err != nil { return nil, err } return c.gg.compile(ctx, option) } // addEndIfNeeded add END edge of the chain/graph. // only run once when compiling. func (c *Chain[I, O]) addEndIfNeeded() error { if c.hasEnd { return nil } if c.err != nil { return c.err } if len(c.preNodeKeys) == 0 { return fmt.Errorf("pre node keys not set, number of nodes in chain= %d", len(c.gg.nodes)) } for _, nodeKey := range c.preNodeKeys { err := c.gg.AddEdge(nodeKey, END) if err != nil { return err } } c.hasEnd = true return nil } func (c *Chain[I, O]) getGenericHelper() *genericHelper { return newGenericHelper[I, O]() } // inputType returns the input type of the chain. // implements AnyGraph. func (c *Chain[I, O]) inputType() reflect.Type { return generic.TypeOf[I]() } // outputType returns the output type of the chain. // implements AnyGraph. func (c *Chain[I, O]) outputType() reflect.Type { return generic.TypeOf[O]() } // compositeType returns the composite type of the chain. // implements AnyGraph. func (c *Chain[I, O]) component() component { return c.gg.component() } // Compile to a Runnable. // Runnable can be used directly. // e.g. // // chain := NewChain[string, string]() // r, err := chain.Compile() // if err != nil {} // // r.Invoke(ctx, input) // ping => pong // r.Stream(ctx, input) // ping => stream out // r.Collect(ctx, inputReader) // stream in => pong // r.Transform(ctx, inputReader) // stream in => stream out func (c *Chain[I, O]) Compile(ctx context.Context, opts ...GraphCompileOption) (Runnable[I, O], error) { if err := c.addEndIfNeeded(); err != nil { return nil, err } return c.gg.Compile(ctx, opts...) } // AppendChatModel add a ChatModel node to the chain. // e.g. // // model, err := openai.NewChatModel(ctx, config) // if err != nil {...} // chain.AppendChatModel(model) func (c *Chain[I, O]) AppendChatModel(node model.BaseChatModel, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toChatModelNode(node, opts...) c.addNode(gNode, options) return c } // AppendChatTemplate add a ChatTemplate node to the chain. // eg. // // chatTemplate, err := prompt.FromMessages(schema.FString, &schema.Message{ // Role: schema.System, // Content: "You are acting as a {role}.", // }) // // chain.AppendChatTemplate(chatTemplate) func (c *Chain[I, O]) AppendChatTemplate(node prompt.ChatTemplate, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toChatTemplateNode(node, opts...) c.addNode(gNode, options) return c } // AppendToolsNode add a ToolsNode node to the chain. // e.g. // // toolsNode, err := tools.NewToolNode(ctx, &tools.ToolsNodeConfig{ // Tools: []tools.Tool{...}, // }) // // chain.AppendToolsNode(toolsNode) func (c *Chain[I, O]) AppendToolsNode(node *ToolsNode, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toToolsNode(node, opts...) c.addNode(gNode, options) return c } // AppendDocumentTransformer add a DocumentTransformer node to the chain. // e.g. // // markdownSplitter, err := markdown.NewHeaderSplitter(ctx, &markdown.HeaderSplitterConfig{}) // // chain.AppendDocumentTransformer(markdownSplitter) func (c *Chain[I, O]) AppendDocumentTransformer(node document.Transformer, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toDocumentTransformerNode(node, opts...) c.addNode(gNode, options) return c } // AppendLambda add a Lambda node to the chain. // Lambda is a node that can be used to implement custom logic. // e.g. // // lambdaNode := compose.InvokableLambda(func(ctx context.Context, docs []*schema.Document) (string, error) {...}) // chain.AppendLambda(lambdaNode) // // Note: // to create a Lambda node, you need to use `compose.AnyLambda` or `compose.InvokableLambda` or `compose.StreamableLambda` or `compose.TransformableLambda`. // if you want this node has real stream output, you need to use `compose.StreamableLambda` or `compose.TransformableLambda`, for example. func (c *Chain[I, O]) AppendLambda(node *Lambda, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toLambdaNode(node, opts...) c.addNode(gNode, options) return c } // AppendEmbedding add a Embedding node to the chain. // e.g. // // embedder, err := openai.NewEmbedder(ctx, config) // if err != nil {...} // chain.AppendEmbedding(embedder) func (c *Chain[I, O]) AppendEmbedding(node embedding.Embedder, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toEmbeddingNode(node, opts...) c.addNode(gNode, options) return c } // AppendRetriever add a Retriever node to the chain. // e.g. // // retriever, err := vectorstore.NewRetriever(ctx, config) // if err != nil {...} // chain.AppendRetriever(retriever) // // or using fornax knowledge as retriever: // // config := fornaxknowledge.Config{...} // retriever, err := fornaxknowledge.NewKnowledgeRetriever(ctx, config) // if err != nil {...} // chain.AppendRetriever(retriever) func (c *Chain[I, O]) AppendRetriever(node retriever.Retriever, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toRetrieverNode(node, opts...) c.addNode(gNode, options) return c } // AppendLoader adds a Loader node to the chain. // e.g. // // loader, err := file.NewFileLoader(ctx, &file.FileLoaderConfig{}) // if err != nil {...} // chain.AppendLoader(loader) func (c *Chain[I, O]) AppendLoader(node document.Loader, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toLoaderNode(node, opts...) c.addNode(gNode, options) return c } // AppendIndexer add an Indexer node to the chain. // Indexer is a node that can store documents. // e.g. // // vectorStoreImpl, err := vikingdb.NewVectorStorer(ctx, vikingdbConfig) // in components/vectorstore/vikingdb/vectorstore.go // if err != nil {...} // // config := vectorstore.IndexerConfig{VectorStore: vectorStoreImpl} // indexer, err := vectorstore.NewIndexer(ctx, config) // if err != nil {...} // // chain.AppendIndexer(indexer) func (c *Chain[I, O]) AppendIndexer(node indexer.Indexer, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toIndexerNode(node, opts...) c.addNode(gNode, options) return c } // AppendBranch add a conditional branch to chain. // Each branch within the ChainBranch can be an AnyGraph. // All branches should either lead to END, or converge to another node within the Chain. // e.g. // // cb := compose.NewChainBranch(conditionFunc) // cb.AddChatTemplate("chat_template_key_01", chatTemplate) // cb.AddChatTemplate("chat_template_key_02", chatTemplate2) // chain.AppendBranch(cb) func (c *Chain[I, O]) AppendBranch(b *ChainBranch) *Chain[I, O] { // nolint: byted_s_too_many_lines_in_func if b == nil { c.reportError(fmt.Errorf("append branch invalid, branch is nil")) return c } if b.err != nil { c.reportError(fmt.Errorf("append branch error: %w", b.err)) return c } if len(b.key2BranchNode) == 0 { c.reportError(fmt.Errorf("append branch invalid, nodeList is empty")) return c } if len(b.key2BranchNode) == 1 { c.reportError(fmt.Errorf("append branch invalid, nodeList length = 1")) return c } var startNode string if len(c.preNodeKeys) == 0 { // branch appended directly to START startNode = START } else if len(c.preNodeKeys) == 1 { startNode = c.preNodeKeys[0] } else { c.reportError(fmt.Errorf("append branch invalid, multiple previous nodes: %v ", c.preNodeKeys)) return c } prefix := c.nextNodeKey() key2NodeKey := make(map[string]string, len(b.key2BranchNode)) for key := range b.key2BranchNode { node := b.key2BranchNode[key] var nodeKey string if node.Second != nil && node.Second.nodeOptions != nil && node.Second.nodeOptions.nodeKey != "" { nodeKey = node.Second.nodeOptions.nodeKey } else { nodeKey = fmt.Sprintf("%s_branch_%s", prefix, key) } if err := c.gg.addNode(nodeKey, node.First, node.Second); err != nil { c.reportError(fmt.Errorf("add branch node[%s] to chain failed: %w", nodeKey, err)) return c } key2NodeKey[key] = nodeKey } gBranch := *b.internalBranch invokeCon := func(ctx context.Context, in any) (endNode []string, err error) { ends, err := b.internalBranch.invoke(ctx, in) if err != nil { return nil, err } nodeKeyEnds := make([]string, 0, len(ends)) for _, end := range ends { if nodeKey, ok := key2NodeKey[end]; !ok { return nil, fmt.Errorf("branch invocation returns unintended end node: %s", end) } else { nodeKeyEnds = append(nodeKeyEnds, nodeKey) } } return nodeKeyEnds, nil } gBranch.invoke = invokeCon collectCon := func(ctx context.Context, sr streamReader) ([]string, error) { ends, err := b.internalBranch.collect(ctx, sr) if err != nil { return nil, err } nodeKeyEnds := make([]string, 0, len(ends)) for _, end := range ends { if nodeKey, ok := key2NodeKey[end]; !ok { return nil, fmt.Errorf("branch invocation returns unintended end node: %s", end) } else { nodeKeyEnds = append(nodeKeyEnds, nodeKey) } } return nodeKeyEnds, nil } gBranch.collect = collectCon gBranch.endNodes = gslice.ToMap(gmap.Values(key2NodeKey), func(k string) (string, bool) { return k, true }) if err := c.gg.AddBranch(startNode, &gBranch); err != nil { c.reportError(fmt.Errorf("chain append branch failed: %w", err)) return c } c.preNodeKeys = gmap.Values(key2NodeKey) return c } // AppendParallel add a Parallel structure (multiple concurrent nodes) to the chain. // e.g. // // parallel := compose.NewParallel() // parallel.AddChatModel("openai", model1) // => "openai": *schema.Message{} // parallel.AddChatModel("maas", model2) // => "maas": *schema.Message{} // // chain.AppendParallel(parallel) // => multiple concurrent nodes are added to the Chain // // The next node in the chain is either an END, or a node which accepts a map[string]any, where keys are `openai` `maas` as specified above. func (c *Chain[I, O]) AppendParallel(p *Parallel) *Chain[I, O] { if p == nil { c.reportError(fmt.Errorf("append parallel invalid, parallel is nil")) return c } if p.err != nil { c.reportError(fmt.Errorf("append parallel invalid, parallel error: %w", p.err)) return c } if len(p.nodes) <= 1 { c.reportError(fmt.Errorf("append parallel invalid, not enough nodes, count = %d", len(p.nodes))) return c } var startNode string if len(c.preNodeKeys) == 0 { // parallel appended directly to START startNode = START } else if len(c.preNodeKeys) == 1 { startNode = c.preNodeKeys[0] } else { c.reportError(fmt.Errorf("append parallel invalid, multiple previous nodes: %v ", c.preNodeKeys)) return c } prefix := c.nextNodeKey() var nodeKeys []string for i := range p.nodes { node := p.nodes[i] var nodeKey string if node.Second != nil && node.Second.nodeOptions != nil && node.Second.nodeOptions.nodeKey != "" { nodeKey = node.Second.nodeOptions.nodeKey } else { nodeKey = fmt.Sprintf("%s_parallel_%d", prefix, i) } if err := c.gg.addNode(nodeKey, node.First, node.Second); err != nil { c.reportError(fmt.Errorf("add parallel node to chain failed, key=%s, err: %w", nodeKey, err)) return c } if err := c.gg.AddEdge(startNode, nodeKey); err != nil { c.reportError(fmt.Errorf("add parallel edge failed, from=%s, to=%s, err: %w", startNode, nodeKey, err)) return c } nodeKeys = append(nodeKeys, nodeKey) } c.preNodeKeys = nodeKeys return c } // AppendGraph add a AnyGraph node to the chain. // AnyGraph can be a chain or a graph. // e.g. // // graph := compose.NewGraph[string, string]() // chain.AppendGraph(graph) func (c *Chain[I, O]) AppendGraph(node AnyGraph, opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toAnyGraphNode(node, opts...) c.addNode(gNode, options) return c } // AppendPassthrough add a Passthrough node to the chain. // Could be used to connect multiple ChainBranch or Parallel. // e.g. // // chain.AppendPassthrough() func (c *Chain[I, O]) AppendPassthrough(opts ...GraphAddNodeOpt) *Chain[I, O] { gNode, options := toPassthroughNode(opts...) c.addNode(gNode, options) return c } // nextIdx. // get the next idx for the chain. // chain key is: node_idx => eg: node_0 => represent the first node of the chain (idx start from 0) // if has parallel: node_idx_parallel_idx => eg: node_0_parallel_1 => represent the first node of the chain, and is a parallel node, and the second node of the parallel // if has branch: node_idx_branch_key => eg: node_1_branch_customkey => represent the second node of the chain, and is a branch node, and the 'customkey' is the key of the branch func (c *Chain[I, O]) nextNodeKey() string { idx := c.nodeIdx c.nodeIdx++ return fmt.Sprintf("node_%d", idx) } // reportError. // save the first error in the chain. func (c *Chain[I, O]) reportError(err error) { if c.err == nil { c.err = err } } // addNode. // add a node to the chain. func (c *Chain[I, O]) addNode(node *graphNode, options *graphAddNodeOpts) { if c.err != nil { return } if c.gg.compiled { c.reportError(ErrChainCompiled) return } if node == nil { c.reportError(fmt.Errorf("chain add node invalid, node is nil")) return } nodeKey := options.nodeOptions.nodeKey defaultNodeKey := c.nextNodeKey() if nodeKey == "" { nodeKey = defaultNodeKey } err := c.gg.addNode(nodeKey, node, options) if err != nil { c.reportError(err) return } if len(c.preNodeKeys) == 0 { c.preNodeKeys = append(c.preNodeKeys, START) } for _, preNodeKey := range c.preNodeKeys { e := c.gg.AddEdge(preNodeKey, nodeKey) if e != nil { c.reportError(e) return } } c.preNodeKeys = []string{nodeKey} } ``` The content has been capped at 50000 tokens, and files over NaN bytes have been omitted. The user could consider applying other filters to refine the result. The better and more specific the context, the better the LLM can follow instructions. If the context seems verbose, the user can refine the filter using uithub. Thank you for using https://uithub.com - Perfect LLM context for any GitHub repo.